Provide DB backed implementation of RegistryStateService (#191)

This mirrors the previous PR which provided a file backed implementation. This will be wired in a future PR.

Note that I changed the DB schema to match the existing sync status structure to simplify implementation. I also added a uniqueness constraint since we now assume that there is only one row in registry_sync per registry.

This PR also renames the LOCAL enum type to MANAGED, and ensures that configurations cannot change the registry type.

Author: dmjb

cmd/thv-registry-api/app/serve_test.go

@@ -191,10 +191,9 @@ func TestRunMigrations(t *testing.T) {
 			err := runMigrations(ctx, cfg)
 			require.NoError(t, err)
 
-			version, dirty, err := database.GetVersion(connStr)
+			_, dirty, err := database.GetVersion(connStr)
 			require.NoError(t, err)
 			require.False(t, dirty)
-			require.Equal(t, uint(1), version)
 		})
 	}
 }

database/migrate_test.go

@@ -27,17 +27,26 @@ func TestMigrations(t *testing.T) {
 	fnames, err := fs.Glob(migrationsFS, "migrations/*.up.sql")
 	require.NoError(t, err)
 
+	// Test each migration individually: apply, rollback, re-apply
 	for i := 1; i <= len(fnames); i++ {
-		// step up
-		err = m.Steps(i)
-		assert.NoError(t, err)
+		// Apply one migration
+		err = m.Steps(1)
+		assert.NoError(t, err, "Failed to apply migration %d", i)
 
-		// step down
-		err = m.Steps(-i)
-		assert.NoError(t, err)
+		// Roll back one migration
+		err = m.Steps(-1)
+		assert.NoError(t, err, "Failed to roll back migration %d", i)
 
-		// step up again
-		err = m.Steps(i)
-		assert.NoError(t, err)
+		// Re-apply the same migration
+		err = m.Steps(1)
+		assert.NoError(t, err, "Failed to re-apply migration %d", i)
 	}
+
+	// Test rolling back all migrations
+	err = m.Down()
+	assert.NoError(t, err, "Failed to roll back all migrations")
+
+	// Test applying all migrations at once
+	err = m.Up()
+	assert.NoError(t, err, "Failed to apply all migrations")
 }

database/migrations/000002_registry_sync_unique_reg_id.down.sql

@@ -0,0 +1,13 @@
+-- Remove additional status fields and unique constraint from registry_sync table
+
+-- Rename MANAGED back to LOCAL in registry_type enum
+ALTER TYPE registry_type RENAME VALUE 'MANAGED' TO 'LOCAL';
+
+-- Remove the additional columns (in reverse order from up migration)
+ALTER TABLE registry_sync DROP COLUMN server_count;
+ALTER TABLE registry_sync DROP COLUMN last_applied_filter_hash;
+ALTER TABLE registry_sync DROP COLUMN last_sync_hash;
+ALTER TABLE registry_sync DROP COLUMN attempt_count;
+
+-- Remove the unique constraint
+ALTER TABLE registry_sync DROP CONSTRAINT registry_sync_reg_id_key;

database/migrations/000002_registry_sync_unique_reg_id.up.sql

@@ -0,0 +1,17 @@
+-- Add unique constraint and additional status fields to registry_sync table
+-- This ensures that each registry can only have one sync status record
+-- and matches the fields in the SyncStatus struct
+
+-- Add unique constraint on reg_id
+ALTER TABLE registry_sync ADD CONSTRAINT registry_sync_reg_id_key UNIQUE (reg_id);
+
+-- Add missing fields from SyncStatus struct
+-- Use BIGINT for count fields to support 64-bit integers
+ALTER TABLE registry_sync ADD COLUMN attempt_count BIGINT NOT NULL DEFAULT 0;
+ALTER TABLE registry_sync ADD COLUMN last_sync_hash TEXT;
+ALTER TABLE registry_sync ADD COLUMN last_applied_filter_hash TEXT;
+ALTER TABLE registry_sync ADD COLUMN server_count BIGINT NOT NULL DEFAULT 0;
+
+-- Rename LOCAL to MANAGED in registry_type enum
+-- Assumping that no prior data exists in the DB at this point.
+ALTER TYPE registry_type RENAME VALUE 'LOCAL' TO 'MANAGED';

database/queries/registry.sql

@@ -46,3 +46,44 @@ INSERT INTO registry (
     sqlc.arg(created_at),
     sqlc.arg(updated_at)
 ) RETURNING id;
+
+-- name: UpsertRegistry :one
+INSERT INTO registry (
+    name,
+    reg_type,
+    created_at,
+    updated_at
+) VALUES (
+    sqlc.arg(name),
+    sqlc.arg(reg_type),
+    sqlc.arg(created_at),
+    sqlc.arg(updated_at)
+)
+ON CONFLICT (name) DO UPDATE SET
+    updated_at = EXCLUDED.updated_at
+RETURNING id;
+
+-- name: BulkUpsertRegistries :many
+INSERT INTO registry (
+    name,
+    reg_type,
+    created_at,
+    updated_at
+)
+SELECT
+    unnest(sqlc.arg(names)::text[]),
+    unnest(sqlc.arg(reg_types)::registry_type[]),
+    unnest(sqlc.arg(created_ats)::timestamp with time zone[]),
+    unnest(sqlc.arg(updated_ats)::timestamp with time zone[])
+ON CONFLICT (name) DO UPDATE SET
+    updated_at = EXCLUDED.updated_at
+RETURNING id, name;
+
+-- name: DeleteRegistriesNotInList :exec
+DELETE FROM registry WHERE id NOT IN (SELECT unnest(sqlc.arg(ids)::uuid[]));
+
+-- name: DeleteRegistry :exec
+DELETE FROM registry WHERE name = sqlc.arg(name);
+
+-- name: ListAllRegistryNames :many
+SELECT name FROM registry ORDER BY name;

database/queries/sync.sql

@@ -4,7 +4,11 @@ SELECT id,
        sync_status,
        error_msg,
        started_at,
-       ended_at
+       ended_at,
+       attempt_count,
+       last_sync_hash,
+       last_applied_filter_hash,
+       server_count
 FROM registry_sync
 WHERE id = sqlc.arg(id);
 
@@ -27,3 +31,90 @@ UPDATE registry_sync SET
     error_msg = sqlc.narg(error_msg),
     ended_at = sqlc.arg(ended_at)
 WHERE id = sqlc.arg(id);
+
+-- name: GetRegistrySyncByName :one
+SELECT rs.id,
+       rs.reg_id,
+       rs.sync_status,
+       rs.error_msg,
+       rs.started_at,
+       rs.ended_at,
+       rs.attempt_count,
+       rs.last_sync_hash,
+       rs.last_applied_filter_hash,
+       rs.server_count
+FROM registry_sync rs
+INNER JOIN registry r ON rs.reg_id = r.id
+WHERE r.name = sqlc.arg(name);
+
+-- name: ListRegistrySyncs :many
+SELECT r.name,
+       rs.id,
+       rs.reg_id,
+       rs.sync_status,
+       rs.error_msg,
+       rs.started_at,
+       rs.ended_at,
+       rs.attempt_count,
+       rs.last_sync_hash,
+       rs.last_applied_filter_hash,
+       rs.server_count
+FROM registry_sync rs
+INNER JOIN registry r ON rs.reg_id = r.id
+ORDER BY rs.started_at DESC, r.name ASC;
+
+-- name: UpsertRegistrySyncByName :exec
+INSERT INTO registry_sync (
+    reg_id,
+    sync_status,
+    error_msg,
+    started_at,
+    ended_at,
+    attempt_count,
+    last_sync_hash,
+    last_applied_filter_hash,
+    server_count
+) VALUES (
+    (SELECT id FROM registry WHERE name = sqlc.arg(name)),
+    sqlc.arg(sync_status),
+    sqlc.narg(error_msg),
+    sqlc.narg(started_at),
+    sqlc.arg(ended_at),
+    sqlc.arg(attempt_count),
+    sqlc.narg(last_sync_hash),
+    sqlc.narg(last_applied_filter_hash),
+    sqlc.arg(server_count)
+)
+ON CONFLICT (reg_id) DO UPDATE SET
+    sync_status = EXCLUDED.sync_status,
+    error_msg = EXCLUDED.error_msg,
+    started_at = EXCLUDED.started_at,
+    ended_at = EXCLUDED.ended_at,
+    attempt_count = EXCLUDED.attempt_count,
+    last_sync_hash = EXCLUDED.last_sync_hash,
+    last_applied_filter_hash = EXCLUDED.last_applied_filter_hash,
+    server_count = EXCLUDED.server_count;
+
+-- name: InitializeRegistrySync :exec
+INSERT INTO registry_sync (
+    reg_id,
+    sync_status,
+    error_msg
+) VALUES (
+    (SELECT id FROM registry WHERE name = sqlc.arg(name)),
+    sqlc.arg(sync_status),
+    sqlc.arg(error_msg)
+)
+ON CONFLICT (reg_id) DO NOTHING;
+
+-- name: BulkInitializeRegistrySyncs :exec
+INSERT INTO registry_sync (
+    reg_id,
+    sync_status,
+    error_msg
+)
+SELECT
+    unnest(sqlc.arg(reg_ids)::uuid[]),
+    unnest(sqlc.arg(sync_statuses)::sync_status[]),
+    unnest(sqlc.arg(error_msgs)::text[])
+ON CONFLICT (reg_id) DO NOTHING;