Add some objects & blob logic to metadata repository

Signed-off-by: Igor Shishkin <me@teran.dev>

repositories/metadata/postgresql/blobs.go

@@ -0,0 +1,26 @@
+package postgresql
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+)
+
+func (r *repository) CreateBLOB(ctx context.Context, checksum string, size uint64, mimeType string) error {
+	_, err := psql.
+		Insert("blobs").
+		Columns(
+			"checksum",
+			"size",
+			"mime_type",
+		).
+		Values(
+			checksum,
+			size,
+			mimeType,
+		).
+		RunWith(r.db).
+		ExecContext(ctx)
+
+	return errors.Wrap(err, "error executing SQL query")
+}

repositories/metadata/postgresql/blobs_test.go

@@ -0,0 +1,6 @@
+package postgresql
+
+func (s *postgreSQLRepositoryTestSuite) TestBlobs() {
+	err := s.repo.CreateBLOB(s.ctx, "deadbeef", 15, "text/plain")
+	s.Require().NoError(err)
+}

repositories/metadata/postgresql/migrations/sql/0000_initial.up.sql

@@ -22,7 +22,7 @@ CREATE TABLE blobs (
 CREATE TABLE objects (
     id SERIAL PRIMARY KEY,
     version_id INT NOT NULL,
-    object_key VARCHAR(255) NOT NULL,
+    key VARCHAR(255) NOT NULL,
     blob_id INT NOT NULL
 );
 
@@ -33,8 +33,10 @@ CREATE UNIQUE INDEX versions_container_id_name_key ON versions (container_id, na
 CREATE INDEX versions_is_published_idx ON versions (is_published);
 
 ALTER TABLE objects ADD FOREIGN KEY (version_id) REFERENCES versions (id);
-CREATE UNIQUE INDEX objects_version_id_object_key_key ON objects (version_id, object_key);
-CREATE INDEX objects_object_key_idx ON objects (object_key);
+CREATE UNIQUE INDEX objects_version_id_key_key ON objects (version_id, key);
+CREATE INDEX objects_key_idx ON objects (key);
 ALTER TABLE objects ADD FOREIGN KEY (blob_id) REFERENCES blobs (id);
 
+CREATE UNIQUE INDEX blobs_checksum_key ON blobs (checksum);
+
 COMMIT;

repositories/metadata/postgresql/objects.go

@@ -0,0 +1,120 @@
+package postgresql
+
+import (
+	"context"
+
+	sq "github.com/Masterminds/squirrel"
+	"github.com/pkg/errors"
+)
+
+func (r *repository) CreateObject(ctx context.Context, container, version, key, casKey string) error {
+	tx, err := r.db.BeginTx(ctx, nil)
+	if err != nil {
+		return errors.Wrap(err, "error beginning transaction")
+	}
+	defer tx.Rollback()
+
+	row := psql.
+		Select("v.id as id").
+		From("containers c").
+		Join("versions v ON v.container_id = c.id").
+		Where(sq.Eq{
+			"c.name":       container,
+			"v.name":       version,
+			"is_published": false,
+		}).
+		RunWith(tx).
+		QueryRowContext(ctx)
+
+	var versionID uint
+	if err := row.Scan(&versionID); err != nil {
+		return errors.Wrap(err, "error looking up version")
+	}
+
+	row = psql.
+		Select("id").
+		From("blobs").
+		Where(sq.Eq{"checksum": casKey}).
+		RunWith(tx).
+		QueryRowContext(ctx)
+
+	var blobID uint
+	if err := row.Scan(&blobID); err != nil {
+		return errors.Wrap(err, "error looking up blob")
+	}
+
+	_, err = psql.
+		Insert("objects").
+		Columns(
+			"version_id",
+			"key",
+			"blob_id",
+		).
+		Values(
+			versionID,
+			key,
+			blobID,
+		).
+		RunWith(tx).
+		ExecContext(ctx)
+	if err != nil {
+		return errors.Wrap(err, "error executing SQL query")
+	}
+
+	if err := tx.Commit(); err != nil {
+		return errors.Wrap(err, "error committing transaction")
+	}
+	return nil
+}
+
+func (r *repository) ListObjects(ctx context.Context, container, version, key string) ([]string, error) {
+	row := psql.
+		Select("v.id").
+		From("versions v").
+		Join("containers c ON v.container_id = c.id").
+		Where(sq.Eq{
+			"c.name": container,
+			"v.name": version,
+		}).
+		RunWith(r.db).
+		QueryRowContext(ctx)
+
+	var versionID uint
+	if err := row.Scan(&versionID); err != nil {
+		return nil, errors.Wrap(err, "error looking up version")
+	}
+
+	rows, err := psql.
+		Select("name").
+		From("objects").
+		Where(sq.Eq{
+			"version_id": versionID,
+		}).
+		OrderBy("id").
+		RunWith(r.db).
+		QueryContext(ctx)
+	if err != nil {
+		return nil, errors.Wrap(err, "error executing SQL query")
+	}
+	defer rows.Close()
+
+	result := []string{}
+	for rows.Next() {
+		var r string
+		if err := rows.Scan(&r); err != nil {
+			return nil, errors.Wrap(err, "error decoding database result")
+		}
+
+		result = append(result, r)
+	}
+
+	return result, nil
+}
+
+func (r *repository) DeleteObject(ctx context.Context, container, version, key string) error {
+	panic("not implemented")
+}
+
+func (r *repository) RemapObject(ctx context.Context, container, version, key, newCASKey string) error {
+	panic("not implemented")
+}

repositories/metadata/postgresql/objects_test.go

@@ -0,0 +1,19 @@
+package postgresql
+
+func (s *postgreSQLRepositoryTestSuite) TestObjects() {
+	const containerName = "test-container-1"
+
+	s.tp.On("Now").Return("2024-07-07T10:11:12Z").Once()
+
+	err := s.repo.CreateContainer(s.ctx, containerName)
+	s.Require().NoError(err)
+
+	versionID, err := s.repo.CreateVersion(s.ctx, containerName)
+	s.Require().NoError(err)
+
+	err = s.repo.CreateBLOB(s.ctx, "deadbeef", 10, "text/plain")
+	s.Require().NoError(err)
+
+	err = s.repo.CreateObject(s.ctx, containerName, versionID, "data/some-key.txt", "deadbeef")
+	s.Require().NoError(err)
+}