Provide DB backed implementation of RegistryStateService

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.

Author: dmjb

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,10 @@
+-- Remove additional status fields and unique constraint from registry_sync table
+
+-- 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,13 @@
+-- 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;

database/queries/registry.sql

@@ -46,3 +46,46 @@ 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
+ reg_type = EXCLUDED.reg_type,
+ 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
+ reg_type = EXCLUDED.reg_type,
+ 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 r.name;
+
+-- 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.arg(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;