chore: ingestion processing cleanup (#209)

Co-authored-by: Michal Fiedorowicz <[email protected]>

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

@@ -26,6 +26,11 @@ WHERE (state = sqlc.narg('state') OR sqlc.narg('state') IS NULL)
 ORDER BY id DESC
 LIMIT sqlc.arg('limit') OFFSET sqlc.arg('offset');
 
+-- name: RetrieveIngestionLogByExternalID :one
+SELECT *
+FROM ingestion_logs
+WHERE external_id = $1;
+
 -- name: RetrieveIngestionLogsWithChangeSets :many
 SELECT v_ingestion_logs_with_change_set.*
 FROM v_ingestion_logs_with_change_set

diode-server/dbstore/postgres/repository.go

@@ -56,6 +56,19 @@ func (r *Repository) CreateIngestionLog(ctx context.Context, ingestionLog *recon
 	return &createdIngestionLog.ID, nil
 }
 
+// RetrieveIngestionLogByExternalID looks up an ingestion log using its external identifier (uuid)
+func (r *Repository) RetrieveIngestionLogByExternalID(ctx context.Context, uuid string) (*int32, *reconcilerpb.IngestionLog, error) {
+	ingestionLog, err := r.queries.RetrieveIngestionLogByExternalID(ctx, uuid)
+	if err != nil {
+		return nil, nil, err
+	}
+	log, err := ingestionLog.ToProto()
+	if err != nil {
+		return nil, nil, err
+	}
+	return &ingestionLog.ID, log, nil
+}
+
 // UpdateIngestionLogStateWithError updates an ingestion log with a new state and error.
 func (r *Repository) UpdateIngestionLogStateWithError(ctx context.Context, id int32, state reconcilerpb.State, ingestionError *reconcilerpb.IngestionError) error {
 	params := postgres.UpdateIngestionLogStateWithErrorParams{

diode-server/gen/dbstore/postgres/adapters.go

@@ -0,0 +1,40 @@
+package postgres
+
+import (
+	"fmt"
+
+	"google.golang.org/protobuf/encoding/protojson"
+
+	"github.com/netboxlabs/diode/diode-server/gen/diode/v1/diodepb"
+	"github.com/netboxlabs/diode/diode-server/gen/diode/v1/reconcilerpb"
+)
+
+// ToProto converts sqlc structure to analogous protobuf
+func (log IngestionLog) ToProto() (*reconcilerpb.IngestionLog, error) {
+	entity := &diodepb.Entity{}
+	if err := protojson.Unmarshal(log.Entity, entity); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal entity: %w", err)
+	}
+	var ingestionErr reconcilerpb.IngestionError
+	if log.Error != nil {
+		if err := protojson.Unmarshal(log.Error, &ingestionErr); err != nil {
+			return nil, fmt.Errorf("failed to unmarshal error: %w", err)
+		}
+	}
+
+	pblog := &reconcilerpb.IngestionLog{
+		Id:                 log.ExternalID,
+		DataType:           log.DataType.String,
+		State:              reconcilerpb.State(log.State.Int32),
+		RequestId:          log.RequestID.String,
+		IngestionTs:        log.IngestionTs.Int64,
+		ProducerAppName:    log.ProducerAppName.String,
+		ProducerAppVersion: log.ProducerAppVersion.String,
+		SdkName:            log.SdkName.String,
+		SdkVersion:         log.SdkVersion.String,
+		Entity:             entity,
+		Error:              &ingestionErr,
+	}
+
+	return pblog, nil
+}