feat: initial commit

pkg/storage/sqlite/blob_bucket.go

@@ -0,0 +1,424 @@
+package sqlite
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"io"
+	"sync"
+	"time"
+
+	"forge.cadoles.com/arcad/edge/pkg/storage"
+	"github.com/gabriel-vasile/mimetype"
+	"github.com/pkg/errors"
+	"gitlab.com/wpetit/goweb/logger"
+)
+
+type BlobBucket struct {
+	name   string
+	getDB  getDBFunc
+	closed bool
+}
+
+// Size implements storage.BlobBucket
+func (b *BlobBucket) Size(ctx context.Context) (int64, error) {
+	var size int64
+
+	err := b.withTx(ctx, func(tx *sql.Tx) error {
+		query := `SELECT SUM(size) FROM blobs WHERE bucket = $1`
+
+		row := tx.QueryRowContext(ctx, query, b.name)
+
+		var nullSize sql.NullInt64
+
+		if err := row.Scan(&nullSize); err != nil {
+			return errors.WithStack(err)
+		}
+
+		size = nullSize.Int64
+
+		return nil
+	})
+	if err != nil {
+		return 0, errors.WithStack(err)
+	}
+
+	return size, nil
+}
+
+// Name implements storage.BlobBucket
+func (b *BlobBucket) Name() string {
+	return b.name
+}
+
+// Close implements storage.BlobBucket
+func (b *BlobBucket) Close() error {
+	logger.Debug(
+		context.Background(), "closing bucket",
+		logger.F("alreadyClosed", b.closed),
+		logger.F("name", b.name),
+	)
+
+	b.closed = true
+
+	return nil
+}
+
+// Delete implements storage.BlobBucket
+func (b *BlobBucket) Delete(ctx context.Context, id storage.BlobID) error {
+	err := b.withTx(ctx, func(tx *sql.Tx) error {
+		query := `DELETE FROM blobs WHERE bucket = $1 AND id = $2`
+
+		if _, err := tx.ExecContext(ctx, query, b.name, id); err != nil {
+			return errors.WithStack(err)
+		}
+
+		return nil
+	})
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
+// Get implements storage.BlobBucket
+func (b *BlobBucket) Get(ctx context.Context, id storage.BlobID) (storage.BlobInfo, error) {
+	var blobInfo *BlobInfo
+
+	err := b.withTx(ctx, func(tx *sql.Tx) error {
+		query := `SELECT content_type, mod_time, size FROM blobs WHERE bucket = $1 AND id = $2`
+		row := tx.QueryRowContext(ctx, query, b.name, id)
+
+		var (
+			contentType string
+			modTime     time.Time
+			size        int64
+		)
+
+		if err := row.Scan(&contentType, &modTime, &size); err != nil {
+			if errors.Is(err, sql.ErrNoRows) {
+				return errors.WithStack(storage.ErrBlobNotFound)
+			}
+
+			return errors.WithStack(err)
+		}
+
+		blobInfo = &BlobInfo{
+			id:          id,
+			bucket:      b.name,
+			contentType: contentType,
+			modTime:     modTime,
+			size:        size,
+		}
+
+		return nil
+	})
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return blobInfo, nil
+}
+
+// List implements storage.BlobBucket
+func (b *BlobBucket) List(ctx context.Context) ([]storage.BlobInfo, error) {
+	var blobs []storage.BlobInfo
+
+	err := b.withTx(ctx, func(tx *sql.Tx) error {
+		query := `SELECT id, content_type, mod_time, size FROM blobs WHERE bucket = $1`
+
+		rows, err := tx.QueryContext(ctx, query, b.name)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+
+		blobs = make([]storage.BlobInfo, 0)
+
+		for rows.Next() {
+			var (
+				blobID      string
+				contentType string
+				modTime     time.Time
+				size        int64
+			)
+
+			if err := rows.Scan(&blobID, &contentType, &modTime, &size); err != nil {
+				if errors.Is(err, sql.ErrNoRows) {
+					return errors.WithStack(storage.ErrBlobNotFound)
+				}
+
+				return errors.WithStack(err)
+			}
+
+			blobInfo := &BlobInfo{
+				id:          storage.BlobID(blobID),
+				bucket:      b.name,
+				contentType: contentType,
+				modTime:     modTime,
+				size:        size,
+			}
+
+			blobs = append(blobs, blobInfo)
+		}
+
+		if err := rows.Err(); err != nil {
+			return errors.WithStack(err)
+		}
+
+		return nil
+	})
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return blobs, nil
+}
+
+// NewReader implements storage.BlobBucket
+func (b *BlobBucket) NewReader(ctx context.Context, id storage.BlobID) (io.ReadSeekCloser, error) {
+	if b.closed {
+		return nil, errors.WithStack(storage.ErrBucketClosed)
+	}
+
+	return &blobReaderCloser{
+		id:     id,
+		bucket: b.name,
+		getDB:  b.getDB,
+	}, nil
+}
+
+// NewWriter implements storage.BlobBucket
+func (b *BlobBucket) NewWriter(ctx context.Context, id storage.BlobID) (io.WriteCloser, error) {
+	if b.closed {
+		return nil, errors.WithStack(storage.ErrBucketClosed)
+	}
+
+	return &blobWriterCloser{
+		id:     id,
+		bucket: b.name,
+		getDB:  b.getDB,
+		buf:    bytes.Buffer{},
+	}, nil
+}
+
+func (b *BlobBucket) withTx(ctx context.Context, fn func(tx *sql.Tx) error) error {
+	if b.closed {
+		return errors.WithStack(storage.ErrBucketClosed)
+	}
+
+	db, err := b.getDB(ctx)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	if err := withTx(ctx, db, fn); err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
+type blobWriterCloser struct {
+	id     storage.BlobID
+	bucket string
+	getDB  getDBFunc
+	buf    bytes.Buffer
+	closed bool
+}
+
+// Write implements io.WriteCloser
+func (wbc *blobWriterCloser) Write(p []byte) (int, error) {
+	logger.Debug(context.Background(), "writing data to blob", logger.F("data", p))
+
+	n, err := wbc.buf.Write(p)
+	if err != nil {
+		return n, errors.WithStack(err)
+	}
+
+	return n, nil
+}
+
+// Close implements io.WriteCloser
+func (wbc *blobWriterCloser) Close() error {
+	ctx := context.Background()
+
+	logger.Debug(
+		ctx, "closing writer",
+		logger.F("alreadyClosed", wbc.closed),
+		logger.F("bucket", wbc.bucket),
+		logger.F("blobID", wbc.id),
+	)
+
+	if wbc.closed {
+		return nil
+	}
+
+	err := wbc.withTx(ctx, func(tx *sql.Tx) error {
+		query := `
+			INSERT INTO blobs (bucket, id, data, content_type, mod_time, size)
+			VALUES ($1, $2, $3, $4, $5, $6)
+			ON CONFLICT (id, bucket) DO UPDATE SET
+			data = $3, content_type = $4, mod_time = $5, size = $6
+		`
+
+		data := wbc.buf.Bytes()
+		mime := mimetype.Detect(data)
+		modTime := time.Now().UTC()
+
+		_, err := tx.Exec(
+			query,
+			wbc.bucket,
+			wbc.id,
+			data,
+			mime.String(),
+			modTime,
+			len(data),
+		)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+
+		return nil
+	})
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	wbc.closed = true
+
+	return nil
+}
+
+func (wbc *blobWriterCloser) withTx(ctx context.Context, fn func(tx *sql.Tx) error) error {
+	if wbc.closed {
+		return errors.WithStack(io.ErrClosedPipe)
+	}
+
+	db, err := wbc.getDB(ctx)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	if err := withTx(ctx, db, fn); err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
+type blobReaderCloser struct {
+	id     storage.BlobID
+	bucket string
+	getDB  getDBFunc
+	reader bytes.Reader
+	once   sync.Once
+	closed bool
+}
+
+// Read implements io.ReadSeekCloser
+func (brc *blobReaderCloser) Read(p []byte) (int, error) {
+	var err error
+
+	brc.once.Do(func() {
+		err = brc.loadBlob()
+	})
+
+	if err != nil {
+		return 0, errors.WithStack(err)
+	}
+
+	n, err := brc.reader.Read(p)
+	if err != nil {
+		if errors.Is(err, io.EOF) {
+			return n, io.EOF
+		}
+
+		return n, errors.WithStack(err)
+	}
+
+	return n, nil
+}
+
+// Seek implements io.ReadSeekCloser
+func (brc *blobReaderCloser) Seek(offset int64, whence int) (int64, error) {
+	var err error
+
+	brc.once.Do(func() {
+		err = brc.loadBlob()
+	})
+
+	if err != nil {
+		return 0, errors.WithStack(err)
+	}
+
+	n, err := brc.reader.Seek(offset, whence)
+	if err != nil {
+		return n, errors.WithStack(err)
+	}
+
+	return n, nil
+}
+
+func (brc *blobReaderCloser) loadBlob() error {
+	ctx := context.Background()
+	logger.Debug(ctx, "loading blob", logger.F("alreadyClosed", brc.closed))
+
+	err := brc.withTx(ctx, func(tx *sql.Tx) error {
+		query := `SELECT data FROM blobs WHERE bucket = $1 AND id = $2`
+		row := tx.QueryRow(query, brc.bucket, brc.id)
+
+		var data []byte
+
+		if err := row.Scan(&data); err != nil {
+			if errors.Is(err, sql.ErrNoRows) {
+				return errors.WithStack(storage.ErrBlobNotFound)
+			}
+
+			return errors.WithStack(err)
+		}
+
+		brc.reader = *bytes.NewReader(data)
+
+		return nil
+	})
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
+// Close implements io.ReadSeekCloser
+func (brc *blobReaderCloser) Close() error {
+	logger.Debug(
+		context.Background(), "closing reader",
+		logger.F("alreadyClosed", brc.closed),
+		logger.F("bucket", brc.bucket),
+		logger.F("blobID", brc.id),
+	)
+
+	brc.closed = true
+
+	return nil
+}
+
+func (brc *blobReaderCloser) withTx(ctx context.Context, fn func(tx *sql.Tx) error) error {
+	db, err := brc.getDB(ctx)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	if err := withTx(ctx, db, fn); err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
+var (
+	_ storage.BlobBucket = &BlobBucket{}
+	_ storage.BlobInfo   = &BlobInfo{}
+	_ io.WriteCloser     = &blobWriterCloser{}
+	_ io.ReadSeekCloser  = &blobReaderCloser{}
+)

pkg/storage/sqlite/blob_info.go

@@ -0,0 +1,40 @@
+package sqlite
+
+import (
+	"time"
+
+	"forge.cadoles.com/arcad/edge/pkg/storage"
+)
+
+type BlobInfo struct {
+	id          storage.BlobID
+	bucket      string
+	contentType string
+	modTime     time.Time
+	size        int64
+}
+
+// Bucket implements storage.BlobInfo
+func (i *BlobInfo) Bucket() string {
+	return i.bucket
+}
+
+// ID implements storage.BlobInfo
+func (i *BlobInfo) ID() storage.BlobID {
+	return i.id
+}
+
+// ContentType implements storage.BlobInfo
+func (i *BlobInfo) ContentType() string {
+	return i.contentType
+}
+
+// ModTime implements storage.BlobInfo
+func (i *BlobInfo) ModTime() time.Time {
+	return i.modTime
+}
+
+// Size implements storage.BlobInfo
+func (i *BlobInfo) Size() int64 {
+	return i.size
+}

pkg/storage/sqlite/blob_store.go

@@ -0,0 +1,136 @@
+package sqlite
+
+import (
+	"context"
+	"database/sql"
+
+	"forge.cadoles.com/arcad/edge/pkg/storage"
+	"github.com/pkg/errors"
+	"gitlab.com/wpetit/goweb/logger"
+)
+
+type BlobStore struct {
+	getDB getDBFunc
+}
+
+// DeleteBucket implements storage.BlobStore
+func (s *BlobStore) DeleteBucket(ctx context.Context, name string) error {
+	err := s.withTx(ctx, func(tx *sql.Tx) error {
+		query := `DELETE FROM blobs WHERE bucket = $1`
+		_, err := tx.ExecContext(ctx, query, name)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+
+		return nil
+	})
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
+// ListBuckets implements storage.BlobStore
+func (s *BlobStore) ListBuckets(ctx context.Context) ([]string, error) {
+	buckets := make([]string, 0)
+
+	err := s.withTx(ctx, func(tx *sql.Tx) error {
+		query := `SELECT DISTINCT name FROM blobs`
+		rows, err := tx.QueryContext(ctx, query)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+
+		defer func() {
+			if err := rows.Close(); err != nil {
+				logger.Error(ctx, "could not close rows", logger.E(errors.WithStack(err)))
+			}
+		}()
+
+		for rows.Next() {
+			var name string
+			if err := rows.Scan(&name); err != nil {
+				return errors.WithStack(err)
+			}
+
+			buckets = append(buckets, name)
+		}
+
+		if err := rows.Err(); err != nil {
+			return errors.WithStack(err)
+		}
+
+		return nil
+	})
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return buckets, nil
+}
+
+// OpenBucket implements storage.BlobStore
+func (s *BlobStore) OpenBucket(ctx context.Context, name string) (storage.BlobBucket, error) {
+	return &BlobBucket{
+		name:  name,
+		getDB: s.getDB,
+	}, nil
+}
+
+func ensureBlobTables(ctx context.Context, db *sql.DB) error {
+	logger.Debug(ctx, "creating blobs table")
+
+	err := withTx(ctx, db, func(tx *sql.Tx) error {
+		query := `
+			CREATE TABLE IF NOT EXISTS blobs (
+				id TEXT,
+				bucket TEXT,
+				data BLOB,
+				content_type TEXT NOT NULL,
+				mod_time TIMESTAMP NOT NULL,
+				size INTEGER,
+				PRIMARY KEY (id, bucket)
+			);
+		`
+		if _, err := tx.ExecContext(ctx, query); err != nil {
+			return errors.WithStack(err)
+		}
+
+		return nil
+	})
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
+func (s *BlobStore) withTx(ctx context.Context, fn func(tx *sql.Tx) error) error {
+	var db *sql.DB
+
+	db, err := s.getDB(ctx)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	if err := withTx(ctx, db, fn); err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
+func NewBlobStore(dsn string) *BlobStore {
+	getDB := newGetDBFunc(dsn, ensureBlobTables)
+
+	return &BlobStore{getDB}
+}
+
+func NewBlobStoreWithDB(db *sql.DB) *BlobStore {
+	getDB := newGetDBFuncFromDB(db, ensureBlobTables)
+
+	return &BlobStore{getDB}
+}
+
+var _ storage.BlobStore = &BlobStore{}

pkg/storage/sqlite/blob_store_test.go

@@ -0,0 +1,25 @@
+package sqlite
+
+import (
+	"os"
+	"testing"
+
+	"forge.cadoles.com/arcad/edge/pkg/storage/testsuite"
+	"github.com/pkg/errors"
+	"gitlab.com/wpetit/goweb/logger"
+)
+
+func TestBlobStore(t *testing.T) {
+	t.Parallel()
+	logger.SetLevel(logger.LevelDebug)
+
+	file := "./testdata/blobstore_test.sqlite"
+
+	if err := os.Remove(file); err != nil && !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("%+v", errors.WithStack(err))
+	}
+
+	store := NewBlobStore(file)
+
+	testsuite.TestBlobStore(t, store)
+}

pkg/storage/sqlite/document_store.go

@@ -0,0 +1,340 @@
+package sqlite
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"sync"
+	"time"
+
+	"forge.cadoles.com/arcad/edge/pkg/storage"
+	"forge.cadoles.com/arcad/edge/pkg/storage/filter"
+	filterSQL "forge.cadoles.com/arcad/edge/pkg/storage/filter/sql"
+	"github.com/pkg/errors"
+	"gitlab.com/wpetit/goweb/logger"
+
+	_ "modernc.org/sqlite"
+)
+
+type DocumentStore struct {
+	db       *sql.DB
+	path     string
+	openOnce sync.Once
+	mutex    sync.RWMutex
+}
+
+// Delete implements storage.DocumentStore
+func (s *DocumentStore) Delete(ctx context.Context, collection string, id storage.DocumentID) error {
+	err := s.withTx(ctx, func(tx *sql.Tx) error {
+		query := `
+		DELETE FROM documents
+		WHERE collection = $1 AND id = $2 
+		`
+
+		_, err := tx.ExecContext(ctx, query, collection, string(id))
+		if err != nil {
+			return errors.WithStack(err)
+		}
+
+		return nil
+	})
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
+// Get implements storage.DocumentStore
+func (s *DocumentStore) Get(ctx context.Context, collection string, id storage.DocumentID) (storage.Document, error) {
+	var document storage.Document
+
+	err := s.withTx(ctx, func(tx *sql.Tx) error {
+		query := `
+		SELECT id, data, created_at, updated_at 
+		FROM documents 
+		WHERE collection = $1 AND id = $2 
+		`
+
+		row := tx.QueryRowContext(ctx, query, collection, string(id))
+
+		var (
+			createdAt time.Time
+			updatedAt time.Time
+			data      JSONMap
+		)
+
+		err := row.Scan(&id, &data, &createdAt, &updatedAt)
+		if err != nil {
+			if errors.Is(err, sql.ErrNoRows) {
+				return errors.WithStack(storage.ErrDocumentNotFound)
+			}
+
+			return errors.WithStack(err)
+		}
+
+		document = storage.Document(data)
+
+		document[storage.DocumentAttrID] = id
+		document[storage.DocumentAttrCreatedAt] = createdAt
+		document[storage.DocumentAttrUpdatedAt] = updatedAt
+
+		return nil
+	})
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return document, nil
+}
+
+// Query implements storage.DocumentStore
+func (s *DocumentStore) Query(ctx context.Context, collection string, filter *filter.Filter, funcs ...storage.QueryOptionFunc) ([]storage.Document, error) {
+	var documents []storage.Document
+
+	err := s.withTx(ctx, func(tx *sql.Tx) error {
+		criteria, args, err := filterSQL.ToSQL(
+			filter.Root(),
+			filterSQL.WithPreparedParameter("$", 2),
+			filterSQL.WithKeyTransform(func(key string) string {
+				return fmt.Sprintf("json_extract(data, '$.%s')", key)
+			}),
+		)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+
+		query := `
+		SELECT id, data, created_at, updated_at 
+		FROM documents 
+		WHERE collection = $1 AND (` + criteria + `)
+		`
+
+		args = append([]interface{}{collection}, args...)
+
+		logger.Debug(
+			ctx, "executing query",
+			logger.F("query", query),
+			logger.F("args", args),
+		)
+
+		rows, err := tx.QueryContext(ctx, query, args...)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+
+		defer rows.Close()
+
+		documents = make([]storage.Document, 0)
+
+		for rows.Next() {
+			var (
+				id        storage.DocumentID
+				createdAt time.Time
+				updatedAt time.Time
+				data      JSONMap
+			)
+
+			if err := rows.Scan(&id, &data, &createdAt, &updatedAt); err != nil {
+				return errors.WithStack(err)
+			}
+
+			document := storage.Document(data)
+			document[storage.DocumentAttrID] = id
+			document[storage.DocumentAttrCreatedAt] = createdAt
+			document[storage.DocumentAttrUpdatedAt] = updatedAt
+
+			documents = append(documents, document)
+		}
+
+		if err := rows.Err(); err != nil {
+			return errors.WithStack(err)
+		}
+
+		return nil
+	})
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return documents, nil
+}
+
+// Upsert implements storage.DocumentStore
+func (s *DocumentStore) Upsert(ctx context.Context, collection string, document storage.Document) (storage.Document, error) {
+	var upsertedDocument storage.Document
+
+	err := s.withTx(ctx, func(tx *sql.Tx) error {
+		query := `
+		INSERT INTO documents (id, collection, data, created_at, updated_at) 
+		VALUES($1, $2, $3, $4, $4)
+		ON CONFLICT (id, collection) DO UPDATE SET
+			data = $3, updated_at = $4
+		RETURNING "id", "data", "created_at", "updated_at"
+		`
+
+		now := time.Now().UTC()
+
+		id, exists := document.ID()
+		if !exists || id == "" {
+			id = storage.NewDocumentID()
+		}
+
+		delete(document, storage.DocumentAttrID)
+		delete(document, storage.DocumentAttrCreatedAt)
+		delete(document, storage.DocumentAttrUpdatedAt)
+
+		args := []any{id, collection, JSONMap(document), now, now}
+
+		row := tx.QueryRowContext(ctx, query, args...)
+
+		var (
+			createdAt time.Time
+			updatedAt time.Time
+			data      JSONMap
+		)
+
+		err := row.Scan(&id, &data, &createdAt, &updatedAt)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+
+		upsertedDocument = storage.Document(data)
+
+		upsertedDocument[storage.DocumentAttrID] = id
+		upsertedDocument[storage.DocumentAttrCreatedAt] = createdAt
+		upsertedDocument[storage.DocumentAttrUpdatedAt] = updatedAt
+
+		return nil
+	})
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return upsertedDocument, nil
+}
+
+func (s *DocumentStore) withTx(ctx context.Context, fn func(tx *sql.Tx) error) error {
+	var db *sql.DB
+
+	db, err := s.getDatabase(ctx)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	if err := withTx(ctx, db, fn); err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
+func (s *DocumentStore) getDatabase(ctx context.Context) (*sql.DB, error) {
+	s.mutex.RLock()
+	if s.db != nil {
+		defer s.mutex.RUnlock()
+
+		var err error
+
+		s.openOnce.Do(func() {
+			if err = s.ensureTables(ctx, s.db); err != nil {
+				err = errors.WithStack(err)
+
+				return
+			}
+		})
+
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+
+		return s.db, nil
+	}
+
+	s.mutex.RUnlock()
+
+	var (
+		db  *sql.DB
+		err error
+	)
+
+	s.openOnce.Do(func() {
+		db, err = sql.Open("sqlite", s.path)
+		if err != nil {
+			err = errors.WithStack(err)
+
+			return
+		}
+
+		if err = s.ensureTables(ctx, db); err != nil {
+			err = errors.WithStack(err)
+
+			return
+		}
+	})
+
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	if db != nil {
+		s.mutex.Lock()
+		s.db = db
+		s.mutex.Unlock()
+	}
+
+	s.mutex.RLock()
+	defer s.mutex.RUnlock()
+
+	return s.db, nil
+}
+
+func (s *DocumentStore) ensureTables(ctx context.Context, db *sql.DB) error {
+	err := withTx(ctx, db, func(tx *sql.Tx) error {
+		query := `
+			CREATE TABLE IF NOT EXISTS documents (
+				id TEXT PRIMARY KEY,
+				collection TEXT NOT NULL,
+				data TEXT,
+				created_at TIMESTAMP NOT NULL,
+				updated_at TIMESTAMP NOT NULL,
+				UNIQUE(id, collection) ON CONFLICT REPLACE
+			);
+		`
+		if _, err := tx.ExecContext(ctx, query); err != nil {
+			return errors.WithStack(err)
+		}
+
+		query = `
+			CREATE INDEX IF NOT EXISTS collection_idx ON documents (collection);
+		`
+		if _, err := tx.ExecContext(ctx, query); err != nil {
+			return errors.WithStack(err)
+		}
+
+		return nil
+	})
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
+func NewDocumentStore(path string) *DocumentStore {
+	return &DocumentStore{
+		db:       nil,
+		path:     path,
+		openOnce: sync.Once{},
+	}
+}
+
+func NewDocumentStoreWithDB(db *sql.DB) *DocumentStore {
+	return &DocumentStore{
+		db:       db,
+		path:     "",
+		openOnce: sync.Once{},
+	}
+}
+
+var _ storage.DocumentStore = &DocumentStore{}

pkg/storage/sqlite/document_store_test.go

@@ -0,0 +1,25 @@
+package sqlite
+
+import (
+	"os"
+	"testing"
+
+	"forge.cadoles.com/arcad/edge/pkg/storage/testsuite"
+	"github.com/pkg/errors"
+	"gitlab.com/wpetit/goweb/logger"
+)
+
+func TestDocumentStore(t *testing.T) {
+	// t.Parallel()
+	logger.SetLevel(logger.LevelDebug)
+
+	file := "./testdata/documentstore_test.sqlite"
+
+	if err := os.Remove(file); err != nil && !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("%+v", errors.WithStack(err))
+	}
+
+	store := NewDocumentStore(file)
+
+	testsuite.TestDocumentStore(t, store)
+}

pkg/storage/sqlite/json.go

@@ -0,0 +1,42 @@
+package sqlite
+
+import (
+	"database/sql/driver"
+	"encoding/json"
+
+	"github.com/pkg/errors"
+)
+
+type JSONMap map[string]any
+
+func (j *JSONMap) Scan(value interface{}) error {
+	if value == nil {
+		return nil
+	}
+
+	var data []byte
+
+	switch typ := value.(type) {
+	case []byte:
+		data = typ
+	case string:
+		data = []byte(typ)
+	default:
+		return errors.Errorf("unexpected type '%T'", value)
+	}
+
+	if err := json.Unmarshal(data, &j); err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
+func (j JSONMap) Value() (driver.Value, error) {
+	data, err := json.Marshal(j)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return data, nil
+}

pkg/storage/sqlite/sql.go

@@ -0,0 +1,99 @@
+package sqlite
+
+import (
+	"context"
+	"database/sql"
+	"sync"
+
+	"github.com/pkg/errors"
+	"gitlab.com/wpetit/goweb/logger"
+)
+
+func withTx(ctx context.Context, db *sql.DB, fn func(tx *sql.Tx) error) error {
+	var tx *sql.Tx
+
+	tx, err := db.Begin()
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	defer func() {
+		if err := tx.Rollback(); err != nil {
+			if errors.Is(err, sql.ErrTxDone) {
+				return
+			}
+
+			panic(errors.WithStack(err))
+		}
+	}()
+
+	if err = fn(tx); err != nil {
+		return errors.WithStack(err)
+	}
+
+	if err = tx.Commit(); err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
+type getDBFunc func(ctx context.Context) (*sql.DB, error)
+
+func newGetDBFunc(dsn string, initFunc func(ctx context.Context, db *sql.DB) error) getDBFunc {
+	var (
+		db    *sql.DB
+		mutex sync.RWMutex
+	)
+
+	return func(ctx context.Context) (*sql.DB, error) {
+		mutex.RLock()
+		if db != nil {
+			defer mutex.RUnlock()
+
+			return db, nil
+		}
+
+		mutex.RUnlock()
+
+		mutex.Lock()
+		defer mutex.Unlock()
+
+		logger.Debug(ctx, "opening database", logger.F("dsn", dsn))
+
+		newDB, err := sql.Open("sqlite", dsn)
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+
+		logger.Debug(ctx, "initializing database")
+
+		if err = initFunc(ctx, newDB); err != nil {
+			return nil, errors.WithStack(err)
+		}
+
+		db = newDB
+
+		return db, nil
+	}
+}
+
+func newGetDBFuncFromDB(db *sql.DB, initFunc func(ctx context.Context, db *sql.DB) error) getDBFunc {
+	var err error
+
+	initOnce := &sync.Once{}
+
+	return func(ctx context.Context) (*sql.DB, error) {
+		initOnce.Do(func() {
+			logger.Debug(ctx, "initializing database")
+
+			err = initFunc(ctx, db)
+		})
+
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+
+		return db, nil
+	}
+}

pkg/storage/sqlite/testdata/.gitignore

@@ -0,0 +1 @@
+/*.sqlite