[WIP] feat: postgres datastore implementation - part 2

diode-server/cmd/reconciler/main.go

@@ -29,9 +29,11 @@ func main() {
 
 	dbURL := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=disable", cfg.PostgresHost, cfg.PostgresPort, cfg.PostgresUser, cfg.PostgresPassword, cfg.PostgresDBName)
 
-	if err := runDBMigrations(ctx, s.Logger(), dbURL); err != nil {
-		s.Logger().Error("failed to run db migrations", "error", err)
-		os.Exit(1)
+	if cfg.MigrationEnabled {
+		if err := runDBMigrations(ctx, s.Logger(), dbURL); err != nil {
+			s.Logger().Error("failed to run db migrations", "error", err)
+			os.Exit(1)
+		}
 	}
 
 	dbPool, err := pgxpool.New(ctx, dbURL)
@@ -41,10 +43,9 @@ func main() {
 	}
 	defer dbPool.Close()
 
-	ingestionLogRepo := postgres.NewIngestionLogRepository(dbPool)
-	changeSetRepo := postgres.NewChangeSetRepository(dbPool)
+	repository := postgres.NewRepository(dbPool)
 
-	ingestionProcessor, err := reconciler.NewIngestionProcessor(ctx, s.Logger(), ingestionLogRepo, changeSetRepo)
+	ingestionProcessor, err := reconciler.NewIngestionProcessor(ctx, s.Logger(), repository)
 	if err != nil {
 		s.Logger().Error("failed to instantiate ingestion processor", "error", err)
 		os.Exit(1)
@@ -55,7 +56,7 @@ func main() {
 		os.Exit(1)
 	}
 
-	gRPCServer, err := reconciler.NewServer(ctx, s.Logger())
+	gRPCServer, err := reconciler.NewServer(ctx, s.Logger(), repository)
 	if err != nil {
 		s.Logger().Error("failed to instantiate gRPC server", "error", err)
 		os.Exit(1)

diode-server/dbstore/postgres/migrations/00001_ingestion_logs.sql

@@ -3,8 +3,8 @@
 -- Create the ingestion_logs table
 CREATE TABLE IF NOT EXISTS ingestion_logs
 (
-    id                   SERIAL PRIMARY KEY,
-    ingestion_log_ksuid  CHAR(27) NOT NULL,
+    id                   INTEGER PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+    ingestion_log_uuid   VARCHAR(255) NOT NULL,
     data_type            VARCHAR(255),
     state                INTEGER,
     request_id           VARCHAR(255),
@@ -21,10 +21,10 @@ CREATE TABLE IF NOT EXISTS ingestion_logs
 );
 
 -- Create indices
-CREATE INDEX IF NOT EXISTS idx_ingestion_logs_ingestion_log_ksuid ON ingestion_logs(ingestion_log_ksuid);
-CREATE INDEX IF NOT EXISTS idx_ingestion_logs_data_type ON ingestion_logs(data_type);
-CREATE INDEX IF NOT EXISTS idx_ingestion_logs_state ON ingestion_logs(state);
-CREATE INDEX IF NOT EXISTS idx_ingestion_logs_request_id ON ingestion_logs(request_id);
+CREATE INDEX IF NOT EXISTS idx_ingestion_logs_ingestion_log_uuid ON ingestion_logs (ingestion_log_uuid);
+CREATE INDEX IF NOT EXISTS idx_ingestion_logs_data_type ON ingestion_logs (data_type);
+CREATE INDEX IF NOT EXISTS idx_ingestion_logs_state ON ingestion_logs (state);
+CREATE INDEX IF NOT EXISTS idx_ingestion_logs_request_id ON ingestion_logs (request_id);
 
 -- +goose Down
 

diode-server/dbstore/postgres/migrations/00002_change_sets.sql

@@ -3,45 +3,69 @@
 -- Create the change_sets table
 CREATE TABLE IF NOT EXISTS change_sets
 (
-    id SERIAL PRIMARY KEY,
-    change_set_ksuid CHAR(27) NOT NULL,
-    ingestion_log_id INTEGER NOT NULL,
-    branch_name VARCHAR(255),
-    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
-    updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
+    id               INTEGER PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+    change_set_uuid  VARCHAR(255) NOT NULL,
+    ingestion_log_id INTEGER      NOT NULL,
+    branch_id        VARCHAR(255),
+    created_at       TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
+    updated_at       TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
 );
 
 -- Create indices
-CREATE INDEX IF NOT EXISTS idx_change_sets_change_set_ksuid ON change_sets(change_set_ksuid);
+CREATE INDEX IF NOT EXISTS idx_change_sets_change_set_uuid ON change_sets (change_set_uuid);
 
 -- Create the changes table
 CREATE TABLE IF NOT EXISTS changes
 (
-    id SERIAL PRIMARY KEY,
-    change_ksuid CHAR(27) NOT NULL,
-    change_set_id INTEGER NOT NULL,
-    change_type VARCHAR(50) NOT NULL,
-    object_type VARCHAR(100) NOT NULL,
-    object_id INTEGER,
-    object_version INTEGER,
-    data JSONB,
+    id              INTEGER PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+    change_uuid     VARCHAR(255) NOT NULL,
+    change_set_id   INTEGER      NOT NULL,
+    change_type     VARCHAR(50)  NOT NULL,
+    object_type     VARCHAR(100) NOT NULL,
+    object_id       INTEGER,
+    object_version  INTEGER,
+    data            JSONB,
     sequence_number INTEGER,
-    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
-    updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
+    created_at      TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
+    updated_at      TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
 );
 
 -- Create indices
-CREATE INDEX IF NOT EXISTS idx_changes_change_ksuid ON changes(change_ksuid);
-CREATE INDEX IF NOT EXISTS idx_changes_change_set_id ON changes(change_set_id);
-CREATE INDEX IF NOT EXISTS idx_changes_change_type ON changes(change_type);
-CREATE INDEX IF NOT EXISTS idx_changes_object_type ON changes(object_type);
+CREATE INDEX IF NOT EXISTS idx_changes_change_uuid ON changes (change_uuid);
+CREATE INDEX IF NOT EXISTS idx_changes_change_set_id ON changes (change_set_id);
+CREATE INDEX IF NOT EXISTS idx_changes_change_type ON changes (change_type);
+CREATE INDEX IF NOT EXISTS idx_changes_object_type ON changes (object_type);
 
 -- Add foreign key constraints
-ALTER TABLE change_sets ADD CONSTRAINT fk_change_sets_ingestion_logs FOREIGN KEY (ingestion_log_id) REFERENCES ingestion_logs(id);
-ALTER TABLE changes ADD CONSTRAINT fk_changes_change_sets FOREIGN KEY (change_set_id) REFERENCES change_sets(id);
+ALTER TABLE change_sets
+    ADD CONSTRAINT fk_change_sets_ingestion_logs FOREIGN KEY (ingestion_log_id) REFERENCES ingestion_logs (id);
+ALTER TABLE changes
+    ADD CONSTRAINT fk_changes_change_sets FOREIGN KEY (change_set_id) REFERENCES change_sets (id);
+
+-- Create a view to join ingestion_logs with change_sets
+CREATE VIEW v_ingestion_logs_change_sets AS
+(
+SELECT change_sets.*
+FROM ingestion_logs
+         LEFT JOIN change_sets on ingestion_logs.id = change_sets.ingestion_log_id
+    );
+
+-- Create a view to join change_sets with changes
+CREATE VIEW v_change_sets_changes AS
+(
+SELECT changes.*
+FROM change_sets
+         LEFT JOIN changes on change_sets.id = changes.change_set_id
+    );
 
 -- +goose Down
 
+-- Drop the v_ingestion_logs_change_sets view
+DROP VIEW IF EXISTS v_ingestion_logs_change_sets;
+
+-- Drop the v_change_sets_with_changes view
+DROP VIEW IF EXISTS v_change_sets_with_changes;
+
 -- Drop the changes table
 DROP TABLE changes;
 

diode-server/dbstore/postgres/queries/change_sets.sql

@@ -1,12 +1,12 @@
 -- name: CreateChangeSet :one
 
-INSERT INTO change_sets (change_set_ksuid, ingestion_log_id, branch_name)
+INSERT INTO change_sets (change_set_uuid, ingestion_log_id, branch_id)
 VALUES ($1, $2, $3)
 RETURNING *;
 
 -- name: CreateChange :one
 
-INSERT INTO changes (change_ksuid, change_set_id, change_type, object_type, object_id, object_version, data,
+INSERT INTO changes (change_uuid, change_set_id, change_type, object_type, object_id, object_version, data,
                      sequence_number)
 VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
 RETURNING *;

diode-server/dbstore/postgres/queries/ingestion_logs.sql

@@ -1,4 +1,39 @@
 -- name: CreateIngestionLog :one
-INSERT INTO ingestion_logs (ingestion_log_ksuid, data_type, state, request_id, ingestion_ts, producer_app_name,
+INSERT INTO ingestion_logs (ingestion_log_uuid, data_type, state, request_id, ingestion_ts, producer_app_name,
                             producer_app_version, sdk_name, sdk_version, entity, source_metadata)
-VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) RETURNING *;
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
+RETURNING *;
+
+-- name: UpdateIngestionLogStateWithError :exec
+UPDATE ingestion_logs
+SET state = $2,
+    error = $3
+WHERE id = $1
+RETURNING *;
+
+-- name: CountIngestionLogsPerState :many
+SELECT state, COUNT(*) AS count
+FROM ingestion_logs
+GROUP BY state;
+
+-- name: RetrieveIngestionLogs :many
+SELECT *
+FROM ingestion_logs
+WHERE (state = sqlc.narg('state') OR sqlc.narg('state') IS NULL)
+  AND (data_type = sqlc.narg('data_type') OR sqlc.narg('data_type') IS NULL)
+  AND (ingestion_ts >= sqlc.narg('ingestion_ts_start') OR sqlc.narg('ingestion_ts_start') IS NULL)
+  AND (ingestion_ts <= sqlc.narg('ingestion_ts_end') OR sqlc.narg('ingestion_ts_end') IS NULL)
+ORDER BY id DESC
+LIMIT sqlc.arg('limit') OFFSET sqlc.arg('offset');
+
+-- name: RetrieveIngestionLogsWithChangeSets :many
+SELECT sqlc.embed(ingestion_logs), sqlc.embed(v_ingestion_logs_change_sets), sqlc.embed(v_change_sets_changes)
+FROM ingestion_logs
+         LEFT JOIN v_ingestion_logs_change_sets on ingestion_logs.id = v_ingestion_logs_change_sets.ingestion_log_id
+         LEFT JOIN v_change_sets_changes on v_ingestion_logs_change_sets.id = v_change_sets_changes.change_set_id
+WHERE (ingestion_logs.state = sqlc.narg('state') OR sqlc.narg('state') IS NULL)
+  AND (ingestion_logs.data_type = sqlc.narg('data_type') OR sqlc.narg('data_type') IS NULL)
+  AND (ingestion_logs.ingestion_ts >= sqlc.narg('ingestion_ts_start') OR sqlc.narg('ingestion_ts_start') IS NULL)
+  AND (ingestion_logs.ingestion_ts <= sqlc.narg('ingestion_ts_end') OR sqlc.narg('ingestion_ts_end') IS NULL)
+ORDER BY ingestion_logs.id DESC, v_change_sets_changes.sequence_number ASC
+LIMIT sqlc.arg('limit') OFFSET sqlc.arg('offset');