diff --git a/internal/repository/module_snapshot.go b/internal/repository/module_snapshot.go index ffe8aa1..540a04b 100644 --- a/internal/repository/module_snapshot.go +++ b/internal/repository/module_snapshot.go @@ -12,16 +12,24 @@ import ( "github.com/jackc/pgx/v5" ) +func moduleSnapshotRowTableExists(ctx context.Context, q queryRower) bool { + var n int + err := q.QueryRow(ctx, ` + SELECT 1 FROM information_schema.tables + WHERE table_schema = 'public' AND table_name = 'module_prefix_snapshot_row' + LIMIT 1`).Scan(&n) + return err == nil +} + func (p *Postgres) GetModulePrefixSnapshot(tenantID, moduleID string) (*store.ModulePrefixSnapshot, bool, error) { ctx := context.Background() var inputHash string var collectedAt time.Time - var raw []byte err := p.pool.QueryRow(ctx, ` - SELECT input_hash, collected_at, prefixes_json + SELECT input_hash, collected_at FROM module_prefix_snapshot WHERE tenant_id = $1 AND module_id = $2`, - tenantID, moduleID).Scan(&inputHash, &collectedAt, &raw) + tenantID, moduleID).Scan(&inputHash, &collectedAt) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, false, nil @@ -29,9 +37,31 @@ func (p *Postgres) GetModulePrefixSnapshot(tenantID, moduleID string) (*store.Mo return nil, false, err } var prefixes []store.PrefixRow - if len(raw) > 0 { - if err := json.Unmarshal(raw, &prefixes); err != nil { - return nil, false, err + if moduleSnapshotRowTableExists(ctx, p.pool) { + rows, qerr := p.pool.Query(ctx, ` + SELECT prefix::text, community_id::text, source + FROM module_prefix_snapshot_row + WHERE tenant_id = $1::uuid AND module_id = $2::uuid + ORDER BY ord`, tenantID, moduleID) + if qerr != nil { + return nil, false, qerr + } + defer rows.Close() + for rows.Next() { + var pr store.PrefixRow + var comm *string + if err := rows.Scan(&pr.Prefix, &comm, &pr.Source); err != nil { + continue + } + pr.CommunityID = comm + prefixes = append(prefixes, pr) + } + } else { + var raw []byte + if err := p.pool.QueryRow(ctx, ` + SELECT prefixes_json FROM module_prefix_snapshot + WHERE tenant_id = $1 AND module_id = $2`, tenantID, moduleID).Scan(&raw); err == nil && len(raw) > 0 { + _ = json.Unmarshal(raw, &prefixes) } } return &store.ModulePrefixSnapshot{ @@ -45,20 +75,57 @@ func (p *Postgres) SetModulePrefixSnapshot(tenantID, moduleID, inputHash string, if strings.TrimSpace(tenantID) == "" || strings.TrimSpace(moduleID) == "" || strings.TrimSpace(inputHash) == "" { return store.ErrInvalidInput } - raw, err := json.Marshal(prefixes) + ctx := context.Background() + tx, err := p.pool.Begin(ctx) if err != nil { return err } - ctx := context.Background() - _, err = p.pool.Exec(ctx, ` - INSERT INTO module_prefix_snapshot (tenant_id, module_id, input_hash, collected_at, prefixes_json) - VALUES ($1::uuid, $2::uuid, $3, now(), $4::jsonb) + defer func() { _ = tx.Rollback(ctx) }() + _, err = tx.Exec(ctx, ` + INSERT INTO module_prefix_snapshot (tenant_id, module_id, input_hash, collected_at) + VALUES ($1::uuid, $2::uuid, $3, now()) ON CONFLICT (tenant_id, module_id) DO UPDATE SET input_hash = EXCLUDED.input_hash, - collected_at = EXCLUDED.collected_at, - prefixes_json = EXCLUDED.prefixes_json`, - tenantID, moduleID, inputHash, string(raw)) - return err + collected_at = EXCLUDED.collected_at`, + tenantID, moduleID, inputHash) + if err != nil { + return err + } + if moduleSnapshotRowTableExists(ctx, tx) { + if _, err := tx.Exec(ctx, ` + DELETE FROM module_prefix_snapshot_row + WHERE tenant_id = $1::uuid AND module_id = $2::uuid`, tenantID, moduleID); err != nil { + return err + } + for i, pr := range prefixes { + var comm any + if pr.CommunityID != nil && strings.TrimSpace(*pr.CommunityID) != "" { + comm = strings.TrimSpace(*pr.CommunityID) + } + src := pr.Source + if strings.TrimSpace(src) == "" { + src = "render" + } + if _, err := tx.Exec(ctx, ` + INSERT INTO module_prefix_snapshot_row (tenant_id, module_id, ord, prefix, community_id, source) + VALUES ($1::uuid, $2::uuid, $3, $4::cidr, $5::uuid, $6)`, + tenantID, moduleID, i, strings.TrimSpace(pr.Prefix), comm, src); err != nil { + return err + } + } + } else { + raw, err := json.Marshal(prefixes) + if err != nil { + return err + } + if _, err := tx.Exec(ctx, ` + UPDATE module_prefix_snapshot SET prefixes_json = $3::jsonb + WHERE tenant_id = $1::uuid AND module_id = $2::uuid`, + tenantID, moduleID, string(raw)); err != nil { + return err + } + } + return tx.Commit(ctx) } func (p *Postgres) DeleteModulePrefixSnapshot(tenantID, moduleID string) error { diff --git a/migrations/postgres/000019_module_snapshot_rows.down.sql b/migrations/postgres/000019_module_snapshot_rows.down.sql new file mode 100644 index 0000000..13c5814 --- /dev/null +++ b/migrations/postgres/000019_module_snapshot_rows.down.sql @@ -0,0 +1,16 @@ +ALTER TABLE module_prefix_snapshot ADD COLUMN prefixes_json JSONB NOT NULL DEFAULT '[]'; + +UPDATE module_prefix_snapshot mps +SET prefixes_json = COALESCE(( + SELECT jsonb_agg( + jsonb_build_object( + 'prefix', psr.prefix::text, + 'community_id', psr.community_id, + 'source', psr.source + ) ORDER BY psr.ord + ) + FROM module_prefix_snapshot_row psr + WHERE psr.tenant_id = mps.tenant_id AND psr.module_id = mps.module_id +), '[]'::jsonb); + +DROP TABLE IF EXISTS module_prefix_snapshot_row; diff --git a/migrations/postgres/000019_module_snapshot_rows.up.sql b/migrations/postgres/000019_module_snapshot_rows.up.sql new file mode 100644 index 0000000..72415f6 --- /dev/null +++ b/migrations/postgres/000019_module_snapshot_rows.up.sql @@ -0,0 +1,25 @@ +CREATE TABLE module_prefix_snapshot_row ( + tenant_id UUID NOT NULL, + module_id UUID NOT NULL, + ord INTEGER NOT NULL, + prefix CIDR NOT NULL, + community_id UUID, + source TEXT NOT NULL DEFAULT '', + PRIMARY KEY (tenant_id, module_id, ord), + FOREIGN KEY (tenant_id, module_id) + REFERENCES module_prefix_snapshot (tenant_id, module_id) ON DELETE CASCADE +); + +INSERT INTO module_prefix_snapshot_row (tenant_id, module_id, ord, prefix, community_id, source) +SELECT mps.tenant_id, + mps.module_id, + (t.ordinality - 1)::int, + (t.elem->>'prefix')::cidr, + NULLIF(t.elem->>'community_id', '')::uuid, + COALESCE(NULLIF(t.elem->>'source', ''), '') +FROM module_prefix_snapshot mps +CROSS JOIN LATERAL jsonb_array_elements(mps.prefixes_json) WITH ORDINALITY AS t(elem, ordinality) +WHERE jsonb_typeof(mps.prefixes_json) = 'array' + AND jsonb_array_length(mps.prefixes_json) > 0; + +ALTER TABLE module_prefix_snapshot DROP COLUMN prefixes_json; diff --git a/migrations/sqlite/000019_module_snapshot_rows.down.sql b/migrations/sqlite/000019_module_snapshot_rows.down.sql new file mode 100644 index 0000000..c651313 --- /dev/null +++ b/migrations/sqlite/000019_module_snapshot_rows.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE module_prefix_snapshot ADD COLUMN prefixes_json TEXT NOT NULL DEFAULT '[]'; +DROP TABLE IF EXISTS module_prefix_snapshot_row; diff --git a/migrations/sqlite/000019_module_snapshot_rows.up.sql b/migrations/sqlite/000019_module_snapshot_rows.up.sql new file mode 100644 index 0000000..c4f0b0f --- /dev/null +++ b/migrations/sqlite/000019_module_snapshot_rows.up.sql @@ -0,0 +1,13 @@ +CREATE TABLE module_prefix_snapshot_row ( + tenant_id TEXT NOT NULL, + module_id TEXT NOT NULL, + ord INTEGER NOT NULL, + prefix TEXT NOT NULL, + community_id TEXT, + source TEXT NOT NULL DEFAULT '', + PRIMARY KEY (tenant_id, module_id, ord), + FOREIGN KEY (tenant_id, module_id) + REFERENCES module_prefix_snapshot (tenant_id, module_id) ON DELETE CASCADE +); + +ALTER TABLE module_prefix_snapshot DROP COLUMN prefixes_json;