Compare commits

...
13 Commits
Author SHA1 Message Date
Denozordec 930e42b0b0 fix(db): update prefix handling in module_prefix_snapshot_row
CI / changes (push) Successful in 8s
CI / commitlint (push) Has been skipped
CI / openapi (push) Has been skipped
CI / web (push) Has been skipped
CI / go (push) Successful in 55s
CI / bird2 (push) Successful in 14s
CI / release (push) Successful in 3m21s
Modified the prefix column type in the module_prefix_snapshot_row table to TEXT, allowing for more flexible input. Adjusted related SQL queries and Go struct tags to ensure compatibility with JSON serialization. Cleaned up migration logic to handle prefix and community_id fields more robustly.
2026-05-25 11:16:03 +07:00
Denozordec 16b4923bd7 fix(db): change prefix column type to TEXT in prefix_snapshot_row
CI / changes (push) Successful in 8s
CI / openapi (push) Has been skipped
CI / commitlint (push) Has been skipped
CI / web (push) Has been skipped
CI / go (push) Successful in 54s
CI / bird2 (push) Successful in 14s
CI / release (push) Successful in 3m44s
Updated the prefix column in the prefix_snapshot_row table from CIDR to TEXT to accommodate broader input formats. Adjusted related SQL insert statements accordingly.
2026-05-25 11:08:06 +07:00
DenozordecandCursor 1cfd062835 chore(db): drop unused schema and redundant indexes
CI / changes (push) Successful in 8s
CI / commitlint (push) Has been skipped
CI / openapi (push) Has been skipped
CI / web (push) Has been skipped
CI / go (push) Successful in 54s
CI / bird2 (push) Successful in 14s
CI / release (push) Successful in 3m44s
Удалены мёртвые колонки, fetch_log и избыточные индексы.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-25 10:58:57 +07:00
DenozordecandCursor 21233bd578 chore(db): add job_audit brin index
BRIN-индекс по created_at для retention-удалений.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-25 10:58:57 +07:00
DenozordecandCursor 990cc739df refactor(db): normalize asn prefix cache and ttl cleanup
Строки asn_prefix_cache_row; периодический prune через evobgp-ingest.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-25 10:58:57 +07:00
DenozordecandCursor 3500bd4624 refactor(db): normalize module prefix snapshot rows
Строки префиксов в module_prefix_snapshot_row вместо JSONB blobs.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-25 10:58:56 +07:00
DenozordecandCursor 374575ec01 refactor(db): mark prefix snapshot contract migration
Маркер миграции contract-фазы H2.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-25 10:58:01 +07:00
DenozordecandCursor f57b430052 feat(db): backfill prefix snapshots and switch read path
Backfill миграция; чтение префиксов через snapshot; запись без дублирования rows.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-25 10:58:01 +07:00
DenozordecandCursor 8a9d60a5a7 feat(db): add prefix snapshot tables (expand)
Таблицы prefix_snapshot и prefix_snapshot_row; колонка prefix_snapshot_id.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-25 10:58:01 +07:00
DenozordecandCursor b963311b43 perf(db): keyset pagination for revision prefixes
Keyset по id/ord вместо OFFSET; anti-join для RevisionDiff.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-25 10:56:46 +07:00
DenozordecandCursor 50bdb8232b refactor(db): split revision preview from meta_json
Preview BIRD-фрагменты в config_revision_preview; meta_json только счётчик префиксов.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-25 10:55:52 +07:00
DenozordecandCursor ee8e24ffc6 perf(db): add revision and module query indexes
Индексы для diff/list префиксов и сортировки модулей; covering index для pagination.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-25 10:55:10 +07:00
DenozordecandCursor 44b94caacf docs(db): add postgres schema diagnostics runbook
Runbook для оценки размеров таблиц, seq scan и EXPLAIN перед/после миграций схемы.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-25 10:55:02 +07:00
49 changed files with 1082 additions and 132 deletions
+90
View File
@@ -0,0 +1,90 @@
# Диагностика схемы PostgreSQL (EvoBGP)
Runbook для оценки объёма БД и узких мест **перед** и **после** миграций оптимизации схемы. Выполнять на staging или production read-only сессией.
## Подключение
```bash
psql "$EVOBGP_DATABASE_URL"
```
## 1. Размеры таблиц и индексов
```sql
SELECT relname,
pg_size_pretty(pg_total_relation_size(relid)) AS total,
pg_size_pretty(pg_relation_size(relid)) AS heap,
pg_size_pretty(pg_indexes_size(relid)) AS indexes
FROM pg_catalog.pg_statio_user_tables
ORDER BY pg_total_relation_size(relid) DESC;
```
**Ожидание:** лидеры — `revision_materialized_prefix`, `config_revision` (TOAST от preview), JSONB-кэши.
## 2. Seq scan (горячие таблицы)
```sql
SELECT schemaname, relname, seq_scan, seq_tup_read, idx_scan
FROM pg_stat_user_tables
WHERE schemaname = 'public'
ORDER BY seq_tup_read DESC;
```
Сброс статистики после деплоя: `SELECT pg_stat_reset();` (только осознанно, теряется baseline).
## 3. Неиспользуемые индексы
```sql
SELECT indexrelname, idx_scan, pg_size_pretty(pg_relation_size(indexrelid)) AS size
FROM pg_stat_user_indexes
WHERE schemaname = 'public' AND idx_scan = 0
ORDER BY pg_relation_size(indexrelid) DESC;
```
## 4. Дубликаты в materialized prefixes
Перед UNIQUE `(revision_id, prefix, community_id, source)`:
```sql
SELECT revision_id, prefix, community_id, source, COUNT(*) AS n
FROM revision_materialized_prefix
GROUP BY 1, 2, 3, 4
HAVING COUNT(*) > 1
LIMIT 20;
```
## 5. Шаблон отчёта staging
| Метрика | До | После | Дата |
|---------|-----|-------|------|
| `revision_materialized_prefix` total | | | |
| `config_revision` total | | | |
| `module_prefix_snapshot` total | | | |
| `asn_prefix_cache` total | | | |
| Top seq_scan table | | | |
| Unused indexes (count) | | | |
## 6. EXPLAIN для типовых запросов
```sql
-- Список префиксов ревизии (keyset)
EXPLAIN (ANALYZE, BUFFERS)
SELECT prefix::text, community_id::text, source
FROM revision_materialized_prefix
WHERE revision_id = '<revision-uuid>'::uuid
ORDER BY id
LIMIT 51;
-- Diff added (anti-join)
EXPLAIN (ANALYZE, BUFFERS)
SELECT b.prefix::text
FROM revision_materialized_prefix b
LEFT JOIN revision_materialized_prefix a
ON a.revision_id = '<rev-a>'::uuid AND a.prefix = b.prefix
WHERE b.revision_id = '<rev-b>'::uuid
AND a.prefix IS NULL
ORDER BY b.prefix
LIMIT 5001;
```
Цель: Index Scan / Bitmap Index Scan по `(revision_id, …)`, без Seq Scan на больших таблицах.
+6
View File
@@ -17,6 +17,8 @@ type Deps struct {
Store store.Backend
}
var lastMaintenance time.Time
// Run blocks until ctx is cancelled.
func Run(ctx context.Context, deps *Deps) {
cfg := config.Load()
@@ -34,6 +36,10 @@ func Run(ctx context.Context, deps *Deps) {
log.Printf("evobgp-ingest: stopped")
return
case <-t.C:
if deps.Store != nil && time.Since(lastMaintenance) > time.Hour {
deps.Store.RunPeriodicMaintenance(ctx)
lastMaintenance = time.Now()
}
prefetchCtx, cancel := context.WithTimeout(ctx, 50*time.Second)
err := pipeline.PrefetchCDNSourceETags(prefetchCtx, deps.Store, hc)
cancel()
+61 -14
View File
@@ -11,14 +11,22 @@ import (
"github.com/jackc/pgx/v5"
)
func asnCacheRowTableExists(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 = 'asn_prefix_cache_row'
LIMIT 1`).Scan(&n)
return err == nil
}
func (p *Postgres) GetASNPrefixCache(asn int64) (*store.ASNPrefixCacheEntry, bool, error) {
ctx := context.Background()
var holder string
var fetchedAt time.Time
var raw []byte
err := p.pool.QueryRow(ctx, `
SELECT holder, fetched_at, prefixes_json FROM asn_prefix_cache WHERE asn = $1`, asn).
Scan(&holder, &fetchedAt, &raw)
SELECT holder, fetched_at FROM asn_prefix_cache WHERE asn = $1`, asn).
Scan(&holder, &fetchedAt)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, false, nil
@@ -26,8 +34,25 @@ func (p *Postgres) GetASNPrefixCache(asn int64) (*store.ASNPrefixCacheEntry, boo
return nil, false, err
}
var prefixes []string
if len(raw) > 0 {
_ = json.Unmarshal(raw, &prefixes)
if asnCacheRowTableExists(ctx, p.pool) {
rows, qerr := p.pool.Query(ctx, `
SELECT prefix::text FROM asn_prefix_cache_row WHERE asn = $1 ORDER BY prefix`, asn)
if qerr != nil {
return nil, false, qerr
}
defer rows.Close()
for rows.Next() {
var s string
if err := rows.Scan(&s); err != nil {
continue
}
prefixes = append(prefixes, s)
}
} else {
var raw []byte
if err := p.pool.QueryRow(ctx, `SELECT prefixes_json FROM asn_prefix_cache WHERE asn = $1`, asn).Scan(&raw); err == nil && len(raw) > 0 {
_ = json.Unmarshal(raw, &prefixes)
}
}
return &store.ASNPrefixCacheEntry{
ASN: asn,
@@ -38,18 +63,40 @@ func (p *Postgres) GetASNPrefixCache(asn int64) (*store.ASNPrefixCacheEntry, boo
}
func (p *Postgres) SetASNPrefixCache(asn int64, holder string, prefixes []string) error {
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 asn_prefix_cache (asn, holder, prefixes_json, fetched_at)
VALUES ($1, $2, $3::jsonb, now())
defer func() { _ = tx.Rollback(ctx) }()
_, err = tx.Exec(ctx, `
INSERT INTO asn_prefix_cache (asn, holder, fetched_at)
VALUES ($1, $2, now())
ON CONFLICT (asn) DO UPDATE SET
holder = EXCLUDED.holder,
prefixes_json = EXCLUDED.prefixes_json,
fetched_at = EXCLUDED.fetched_at`,
asn, holder, string(raw))
return err
fetched_at = EXCLUDED.fetched_at`, asn, holder)
if err != nil {
return err
}
if asnCacheRowTableExists(ctx, tx) {
if _, err := tx.Exec(ctx, `DELETE FROM asn_prefix_cache_row WHERE asn = $1`, asn); err != nil {
return err
}
for _, pfx := range prefixes {
if _, err := tx.Exec(ctx, `
INSERT INTO asn_prefix_cache_row (asn, prefix) VALUES ($1, $2::cidr)`, asn, pfx); err != nil {
return err
}
}
} else {
raw, err := json.Marshal(prefixes)
if err != nil {
return err
}
if _, err := tx.Exec(ctx, `
UPDATE asn_prefix_cache SET prefixes_json = $2::jsonb WHERE asn = $1`, asn, string(raw)); err != nil {
return err
}
}
return tx.Commit(ctx)
}
+29
View File
@@ -0,0 +1,29 @@
package repository
import (
"context"
"time"
)
const (
jobAuditRetentionDays = 90
asnCacheRetentionDays = 7
)
// RunPeriodicMaintenance prunes stale job_audit and asn_prefix_cache rows (PostgreSQL).
func (p *Postgres) RunPeriodicMaintenance(ctx context.Context) {
if p == nil || p.pool == nil {
return
}
if ctx == nil {
ctx = context.Background()
}
jobCutoff := time.Now().UTC().Add(-time.Duration(jobAuditRetentionDays) * 24 * time.Hour)
_, _ = p.pool.Exec(ctx, `
DELETE FROM job_audit
WHERE created_at < $1
AND status IN ('succeeded', 'failed', 'cancelled')`, jobCutoff)
asnCutoff := time.Now().UTC().Add(-time.Duration(asnCacheRetentionDays) * 24 * time.Hour)
_, _ = p.pool.Exec(ctx, `
DELETE FROM asn_prefix_cache WHERE fetched_at < $1`, asnCutoff)
}
+82 -15
View File
@@ -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, 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, $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 {
+77 -66
View File
@@ -687,7 +687,7 @@ func (p *Postgres) GetRevision(tenantID, revisionID string) (*store.Revision, er
if mj.PreviewFragments == nil {
mj.PreviewFragments = map[string]string{}
}
r.PreviewFragments = mj.PreviewFragments
r.PreviewFragments = loadRevisionPreview(ctx, p.pool, revisionID, mj.PreviewFragments)
r.MaterializedPrefixCount = mj.MaterializedPrefixCount
return &r, nil
}
@@ -789,12 +789,7 @@ func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor stri
if limit <= 0 {
limit = 50
}
off := 0
if cursor != "" {
if n, err := strconv.Atoi(cursor); err == nil && n >= 0 {
off = n
}
}
afterID, off, useOffset := store.ParsePrefixPageCursor(cursor)
ctx := context.Background()
var one int
if err := p.pool.QueryRow(ctx, `
@@ -805,34 +800,55 @@ func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor stri
}
return nil, "", false
}
rows, err := p.pool.Query(ctx, `
SELECT prefix::text, community_id::text, source FROM revision_materialized_prefix
WHERE revision_id=$1 ORDER BY id
LIMIT $2 OFFSET $3`, revisionID, limit+1, off)
if snapID, ok := p.revisionPrefixSnapshotID(ctx, revisionID); ok {
return p.listSnapshotPrefixes(ctx, snapID, cursor, limit)
}
var rows pgx.Rows
var err error
if useOffset {
rows, err = p.pool.Query(ctx, `
SELECT id, prefix::text, community_id::text, source FROM revision_materialized_prefix
WHERE revision_id=$1::uuid ORDER BY id
LIMIT $2 OFFSET $3`, revisionID, limit+1, off)
} else {
var afterArg any
if afterID != nil {
afterArg = *afterID
}
rows, err = p.pool.Query(ctx, `
SELECT id, prefix::text, community_id::text, source FROM revision_materialized_prefix
WHERE revision_id=$1::uuid AND ($2::bigint IS NULL OR id > $2::bigint)
ORDER BY id
LIMIT $3`, revisionID, afterArg, limit+1)
}
if err != nil {
return nil, "", false
}
defer rows.Close()
var all []store.PrefixRow
var ids []int64
for rows.Next() {
var rowID int64
var pr store.PrefixRow
var comm *string
if err := rows.Scan(&pr.Prefix, &comm, &pr.Source); err != nil {
if err := rows.Scan(&rowID, &pr.Prefix, &comm, &pr.Source); err != nil {
continue
}
pr.CommunityID = comm
ids = append(ids, rowID)
all = append(all, pr)
}
agentDebugNDJSON3214("B", "repository/postgres.go:ListRevisionPrefixes", "list_prefixes_fetched", map[string]any{
"rows": len(all), "limit": limit, "offset": off,
"rows": len(all), "limit": limit, "keyset": !useOffset,
})
more := len(all) > limit
if more {
all = all[:limit]
ids = ids[:limit]
}
next := ""
if more {
next = fmt.Sprintf("%d", off+limit)
if more && len(ids) > 0 {
next = store.FormatPrefixPageCursor(ids[len(ids)-1])
}
if len(all) == 0 {
return nil, "", false
@@ -848,26 +864,35 @@ func (p *Postgres) CreateRollbackRevision(tenantID, sourceRevisionID string) (st
ctx := context.Background()
newID := uuid.NewString()
parent := sourceRevisionID
meta, _ := json.Marshal(map[string]any{
"preview_fragments": src.PreviewFragments,
"materialized_prefix_count": src.MaterializedPrefixCount,
})
meta, err := revisionMetaWithoutPreview(src.MaterializedPrefixCount)
if err != nil {
return "", err
}
var modArg any
if strings.TrimSpace(src.ModuleID) != "" {
modArg = src.ModuleID
}
_, err = p.pool.Exec(ctx, `
INSERT INTO config_revision (id, tenant_id, module_id, content_hash, parent_revision_id, meta_json)
VALUES ($1,$2,$3,$4,$5::uuid,$6::jsonb)`,
newID, tenantID, modArg, src.ContentHash+":rollback", parent, string(meta))
tx, err := p.pool.Begin(ctx)
if err != nil {
return "", err
}
// copy materialized prefixes
_, _ = p.pool.Exec(ctx, `
INSERT INTO revision_materialized_prefix (revision_id, prefix, community_id, source, meta_json)
SELECT $1::uuid, prefix, community_id, source, meta_json FROM revision_materialized_prefix WHERE revision_id=$2::uuid`,
newID, sourceRevisionID)
defer func() { _ = tx.Rollback(ctx) }()
_, err = tx.Exec(ctx, `
INSERT INTO config_revision (id, tenant_id, module_id, content_hash, parent_revision_id, meta_json)
VALUES ($1,$2,$3,$4,$5::uuid,$6::jsonb)`,
newID, tenantID, modArg, src.ContentHash+":rollback", parent, meta)
if err != nil {
return "", err
}
if err := copyRevisionPreview(ctx, tx, newID, sourceRevisionID); err != nil {
return "", err
}
if err := p.copyRevisionPrefixSnapshotRef(ctx, tx, newID, sourceRevisionID); err != nil {
return "", err
}
if err := tx.Commit(ctx); err != nil {
return "", err
}
return newID, nil
}
@@ -884,20 +909,18 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
var unchanged int
err := p.pool.QueryRow(ctx, `
SELECT COUNT(*)::int FROM (
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
INTERSECT
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
SELECT b.prefix FROM (`+sqlRevisionPrefixes("$2")+`) b
INNER JOIN (`+sqlRevisionPrefixes("$1")+`) a ON a.prefix = b.prefix
) t`, aID, bID).Scan(&unchanged)
if err != nil {
return nil, err
}
// added: в B, нет в A; removed: в A, нет в B — без загрузки полных снапшотов в память.
rowsAdded, err := p.pool.Query(ctx, `
SELECT prefix::text FROM (
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
EXCEPT
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
) s ORDER BY 1 LIMIT $3`, bID, aID, maxRevisionDiffRows+1)
SELECT b.prefix::text FROM (`+sqlRevisionPrefixes("$2")+`) b
LEFT JOIN (`+sqlRevisionPrefixes("$1")+`) a ON a.prefix = b.prefix
WHERE a.prefix IS NULL
ORDER BY b.prefix
LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
if err != nil {
return nil, err
}
@@ -916,11 +939,11 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
}
addedTruncated := len(added) >= maxRevisionDiffRows
rowsRem, err := p.pool.Query(ctx, `
SELECT prefix::text FROM (
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
EXCEPT
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
) s ORDER BY 1 LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
SELECT a.prefix::text FROM (`+sqlRevisionPrefixes("$1")+`) a
LEFT JOIN (`+sqlRevisionPrefixes("$2")+`) b ON b.prefix = a.prefix
WHERE b.prefix IS NULL
ORDER BY a.prefix
LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
if err != nil {
return nil, err
}
@@ -1056,10 +1079,7 @@ func (p *Postgres) CreateRenderRevision(revisionID, tenantID, moduleID string, p
if previewFragments == nil {
previewFragments = map[string]string{}
}
meta, err := json.Marshal(map[string]any{
"preview_fragments": previewFragments,
"materialized_prefix_count": len(prefixes),
})
meta, err := revisionMetaWithoutPreview(len(prefixes))
if err != nil {
return err
}
@@ -1073,32 +1093,23 @@ func (p *Postgres) CreateRenderRevision(revisionID, tenantID, moduleID string, p
if parentRevisionID != nil && strings.TrimSpace(*parentRevisionID) != "" {
parent = strings.TrimSpace(*parentRevisionID)
}
revID := strings.TrimSpace(revisionID)
_, err = tx.Exec(ctx, `
INSERT INTO config_revision (id, tenant_id, module_id, content_hash, parent_revision_id, meta_json)
VALUES ($1::uuid, $2::uuid, $3::uuid, $4, $5::uuid, $6::jsonb)`,
strings.TrimSpace(revisionID), tenantID, moduleID, strings.TrimSpace(contentHash), parent, string(meta))
revID, tenantID, moduleID, strings.TrimSpace(contentHash), parent, meta)
if err != nil {
return err
}
if len(prefixes) > 0 {
_, err = tx.CopyFrom(ctx,
pgx.Identifier{"revision_materialized_prefix"},
[]string{"revision_id", "prefix", "community_id", "source"},
pgx.CopyFromSlice(len(prefixes), func(i int) ([]any, error) {
pr := prefixes[i]
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"
}
return []any{strings.TrimSpace(revisionID), strings.TrimSpace(pr.Prefix), comm, src}, nil
}))
if err != nil {
return err
}
if err := insertRevisionPreview(ctx, tx, revID, previewFragments); err != nil {
return err
}
snapID, err := p.ensurePrefixSnapshot(ctx, tx, contentHash, prefixes)
if err != nil {
return err
}
if err := p.linkRevisionPrefixSnapshot(ctx, tx, revID, snapID); err != nil {
return err
}
if err := tx.Commit(ctx); err != nil {
return err
@@ -0,0 +1,186 @@
package repository
import (
"context"
"errors"
"strings"
"evobgp/internal/store"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
)
func prefixSnapshotTableExists(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 = 'prefix_snapshot'
LIMIT 1`).Scan(&n)
return err == nil
}
func normalizeSnapshotHash(contentHash string) string {
h := strings.TrimSpace(contentHash)
if strings.HasPrefix(h, "sha256:") {
h = strings.TrimPrefix(h, "sha256:")
}
if len(h) > 64 {
h = h[:64]
}
if len(h) < 64 {
h = h + strings.Repeat("0", 64-len(h))
}
return h
}
func (p *Postgres) revisionPrefixSnapshotID(ctx context.Context, revisionID string) (string, bool) {
if !prefixSnapshotTableExists(ctx, p.pool) {
return "", false
}
var snap *string
err := p.pool.QueryRow(ctx, `
SELECT prefix_snapshot_id::text FROM config_revision
WHERE id = $1::uuid AND prefix_snapshot_id IS NOT NULL`, revisionID).Scan(&snap)
if err != nil || snap == nil || strings.TrimSpace(*snap) == "" {
return "", false
}
return *snap, true
}
func (p *Postgres) ensurePrefixSnapshot(ctx context.Context, db execQuerier, contentHash string, prefixes []store.PrefixRow) (string, error) {
if !prefixSnapshotTableExists(ctx, db) {
return "", nil
}
hash := normalizeSnapshotHash(contentHash)
var existing string
err := db.QueryRow(ctx, `SELECT id::text FROM prefix_snapshot WHERE content_hash = $1`, hash).Scan(&existing)
if err == nil && existing != "" {
return existing, nil
}
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
return "", err
}
snapID := uuid.NewString()
if _, err := db.Exec(ctx, `
INSERT INTO prefix_snapshot (id, content_hash) VALUES ($1::uuid, $2)
ON CONFLICT (content_hash) DO NOTHING`, snapID, hash); err != nil {
return "", err
}
if err := db.QueryRow(ctx, `SELECT id::text FROM prefix_snapshot WHERE content_hash = $1`, hash).Scan(&snapID); err != nil {
return "", err
}
var rowCount int
_ = db.QueryRow(ctx, `SELECT COUNT(*)::int FROM prefix_snapshot_row WHERE snapshot_id = $1::uuid`, snapID).Scan(&rowCount)
if rowCount > 0 {
return snapID, nil
}
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 := db.Exec(ctx, `
INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source)
VALUES ($1::uuid, $2, $3, $4::uuid, $5)`,
snapID, i, strings.TrimSpace(pr.Prefix), comm, src); err != nil {
return "", err
}
}
return snapID, nil
}
func (p *Postgres) listSnapshotPrefixes(ctx context.Context, snapshotID, cursor string, limit int) ([]store.PrefixRow, string, bool) {
afterOrd, off, useOffset := store.ParsePrefixPageCursor(cursor)
var rows pgx.Rows
var err error
if useOffset {
rows, err = p.pool.Query(ctx, `
SELECT ord, prefix::text, community_id::text, source
FROM prefix_snapshot_row
WHERE snapshot_id = $1::uuid
ORDER BY ord
LIMIT $2 OFFSET $3`, snapshotID, limit+1, off)
} else {
var afterArg any
if afterOrd != nil {
afterArg = int(*afterOrd)
}
rows, err = p.pool.Query(ctx, `
SELECT ord, prefix::text, community_id::text, source
FROM prefix_snapshot_row
WHERE snapshot_id = $1::uuid AND ($2::int IS NULL OR ord > $2::int)
ORDER BY ord
LIMIT $3`, snapshotID, afterArg, limit+1)
}
if err != nil {
return nil, "", false
}
defer rows.Close()
var all []store.PrefixRow
var ords []int64
for rows.Next() {
var ord int
var pr store.PrefixRow
var comm *string
if err := rows.Scan(&ord, &pr.Prefix, &comm, &pr.Source); err != nil {
continue
}
pr.CommunityID = comm
ords = append(ords, int64(ord))
all = append(all, pr)
}
more := len(all) > limit
if more {
all = all[:limit]
ords = ords[:limit]
}
next := ""
if more && len(ords) > 0 {
next = store.FormatPrefixPageCursor(ords[len(ords)-1])
}
if len(all) == 0 {
return nil, "", false
}
return all, next, more
}
func (p *Postgres) linkRevisionPrefixSnapshot(ctx context.Context, db execQuerier, revisionID, snapshotID string) error {
if snapshotID == "" || !prefixSnapshotTableExists(ctx, db) {
return nil
}
_, err := db.Exec(ctx, `
UPDATE config_revision SET prefix_snapshot_id = $2::uuid WHERE id = $1::uuid`,
revisionID, snapshotID)
return err
}
func (p *Postgres) copyRevisionPrefixSnapshotRef(ctx context.Context, db execQuerier, dstRevisionID, srcRevisionID string) error {
if !prefixSnapshotTableExists(ctx, db) {
return nil
}
_, err := db.Exec(ctx, `
UPDATE config_revision dst
SET prefix_snapshot_id = src.prefix_snapshot_id
FROM config_revision src
WHERE dst.id = $1::uuid AND src.id = $2::uuid AND src.prefix_snapshot_id IS NOT NULL`,
dstRevisionID, srcRevisionID)
return err
}
func sqlRevisionPrefixes(revParam string) string {
return `SELECT psr.prefix FROM config_revision cr
JOIN prefix_snapshot_row psr ON psr.snapshot_id = cr.prefix_snapshot_id
WHERE cr.id = ` + revParam + `::uuid AND cr.prefix_snapshot_id IS NOT NULL
UNION ALL
SELECT rmp.prefix FROM revision_materialized_prefix rmp
WHERE rmp.revision_id = ` + revParam + `::uuid
AND NOT EXISTS (
SELECT 1 FROM config_revision cr2
WHERE cr2.id = ` + revParam + `::uuid AND cr2.prefix_snapshot_id IS NOT NULL
)`
}
@@ -0,0 +1,90 @@
package repository
import (
"context"
"encoding/json"
"errors"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgconn"
)
func revisionPreviewTableExists(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 = 'config_revision_preview'
LIMIT 1`).Scan(&n)
return err == nil
}
type queryRower interface {
QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
}
func loadRevisionPreview(ctx context.Context, q queryRower, revisionID string, metaPreview map[string]string) map[string]string {
if revisionPreviewTableExists(ctx, q) {
var raw []byte
err := q.QueryRow(ctx, `
SELECT fragments FROM config_revision_preview WHERE revision_id = $1::uuid`,
revisionID).Scan(&raw)
if err == nil {
out := map[string]string{}
_ = json.Unmarshal(raw, &out)
if out == nil {
out = map[string]string{}
}
return out
}
if !errors.Is(err, pgx.ErrNoRows) {
return metaPreview
}
}
if metaPreview == nil {
return map[string]string{}
}
return metaPreview
}
type execQuerier interface {
queryRower
Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
}
func insertRevisionPreview(ctx context.Context, db execQuerier, revisionID string, preview map[string]string) error {
if !revisionPreviewTableExists(ctx, db) {
return nil
}
if preview == nil {
preview = map[string]string{}
}
raw, err := json.Marshal(preview)
if err != nil {
return err
}
_, err = db.Exec(ctx, `
INSERT INTO config_revision_preview (revision_id, fragments)
VALUES ($1::uuid, $2::jsonb)
ON CONFLICT (revision_id) DO UPDATE SET fragments = EXCLUDED.fragments`,
revisionID, string(raw))
return err
}
func copyRevisionPreview(ctx context.Context, db execQuerier, dstRevisionID, srcRevisionID string) error {
if !revisionPreviewTableExists(ctx, db) {
return nil
}
_, err := db.Exec(ctx, `
INSERT INTO config_revision_preview (revision_id, fragments)
SELECT $1::uuid, fragments FROM config_revision_preview WHERE revision_id = $2::uuid
ON CONFLICT (revision_id) DO UPDATE SET fragments = EXCLUDED.fragments`,
dstRevisionID, srcRevisionID)
return err
}
func revisionMetaWithoutPreview(materializedPrefixCount int) (string, error) {
raw, err := json.Marshal(map[string]any{
"materialized_prefix_count": materializedPrefixCount,
})
return string(raw), err
}
+30 -15
View File
@@ -26,9 +26,8 @@ func (p *Postgres) seedDemo(ctx context.Context) error {
p1 := uuid.NewString()
p2 := uuid.NewString()
preview := map[string]any{
"preview_fragments": map[string]string{
"bird.conf": `# EvoBGP demo bundle
previewFrags := map[string]string{
"bird.conf": `# EvoBGP demo bundle
router id 192.0.2.1;
protocol device {
@@ -39,15 +38,12 @@ protocol direct {
ipv6;
}
`,
"bird.d/evobgp_demo.conf": "# static demo fragment\n",
},
"materialized_prefix_count": 128,
"bird.d/evobgp_demo.conf": "# static demo fragment\n",
}
previewB, _ := json.Marshal(preview)
parentMeta, _ := json.Marshal(map[string]any{
"preview_fragments": map[string]string{"bird.conf": "# parent revision\n"},
"materialized_prefix_count": 0,
})
previewMeta, _ := json.Marshal(map[string]any{"materialized_prefix_count": 128})
parentMeta, _ := json.Marshal(map[string]any{"materialized_prefix_count": 0})
parentPreview, _ := json.Marshal(map[string]string{"bird.conf": "# parent revision\n"})
previewFragsB, _ := json.Marshal(previewFrags)
tx, err := p.pool.Begin(ctx)
if err != nil {
@@ -77,13 +73,32 @@ protocol direct {
return err
}
if _, err := tx.Exec(ctx, `
INSERT INTO config_revision (id, tenant_id, content_hash, parent_revision_id, meta_json)
VALUES ($1,$2,'sha256:demo-rev-1',$3::uuid,$4::jsonb)`, rid, tid, parent, string(previewB)); err != nil {
INSERT INTO config_revision_preview (revision_id, fragments) VALUES ($1::uuid, $2::jsonb)`, parent, string(parentPreview)); err != nil {
return err
}
if _, err := tx.Exec(ctx, `
INSERT INTO revision_materialized_prefix (revision_id, prefix, community_id, source)
VALUES ($1::uuid,'203.0.113.0/24',$2::uuid,'demo'), ($1::uuid,'2001:db8::/32',$2::uuid,'demo')`, rid, cid); err != nil {
INSERT INTO config_revision (id, tenant_id, content_hash, parent_revision_id, meta_json)
VALUES ($1,$2,'sha256:demo-rev-1',$3::uuid,$4::jsonb)`, rid, tid, parent, string(previewMeta)); err != nil {
return err
}
if _, err := tx.Exec(ctx, `
INSERT INTO config_revision_preview (revision_id, fragments) VALUES ($1::uuid, $2::jsonb)`, rid, string(previewFragsB)); err != nil {
return err
}
snapID := uuid.NewString()
demoHash := normalizeSnapshotHash("sha256:demo-rev-1")
if _, err := tx.Exec(ctx, `
INSERT INTO prefix_snapshot (id, content_hash) VALUES ($1::uuid, $2)`, snapID, demoHash); err != nil {
return err
}
if _, err := tx.Exec(ctx, `
UPDATE config_revision SET prefix_snapshot_id = $2::uuid WHERE id = $1::uuid`, rid, snapID); err != nil {
return err
}
if _, err := tx.Exec(ctx, `
INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source)
VALUES ($1::uuid, 0, '203.0.113.0/24', $2::uuid, 'demo'),
($1::uuid, 1, '2001:db8::/32', $2::uuid, 'demo')`, snapID, cid); err != nil {
return err
}
if _, err := tx.Exec(ctx, `
+6 -3
View File
@@ -112,6 +112,9 @@ type Backend interface {
// Ping verifies backend connectivity (no-op for in-memory).
Ping(ctx context.Context) error
// RunPeriodicMaintenance prunes stale DB rows (no-op for in-memory).
RunPeriodicMaintenance(ctx context.Context)
}
// ASNPrefixCacheEntry is a cached RIPEstat response for one ASN.
@@ -311,7 +314,7 @@ type SpeakerPatch struct {
// PrefixRow is one materialized prefix for GET /revisions/.../prefixes.
type PrefixRow struct {
Prefix string
CommunityID *string
Source string
Prefix string `json:"prefix"`
CommunityID *string `json:"community_id,omitempty"`
Source string `json:"source,omitempty"`
}
+5
View File
@@ -305,6 +305,11 @@ func (m *Memory) Ping(ctx context.Context) error {
return nil
}
// RunPeriodicMaintenance is a no-op for the in-memory backend.
func (m *Memory) RunPeriodicMaintenance(ctx context.Context) {
_ = ctx
}
// ListTenantIDs returns tenant ids sorted lexicographically.
func (m *Memory) ListTenantIDs() ([]string, error) {
m.mu.RLock()
+23 -19
View File
@@ -1,8 +1,6 @@
package store
import (
"fmt"
"strconv"
"strings"
"time"
@@ -795,31 +793,37 @@ func (m *Memory) ListRevisionPrefixes(tenantID, revisionID string, cursor string
if limit <= 0 {
limit = 50
}
afterID, off, useOffset := ParsePrefixPageCursor(cursor)
m.mu.RLock()
defer m.mu.RUnlock()
if _, err := m.getRevisionLocked(tenantID, revisionID); err != nil {
return nil, "", false
}
all := m.revPrefixes[revisionID]
off := 0
if cursor != "" {
if n, err := strconv.Atoi(cursor); err == nil && n >= 0 {
off = n
}
allRows := m.revPrefixes[revisionID]
start := 0
if useOffset {
start = off
} else if afterID != nil {
start = int(*afterID) + 1
}
end := off + limit
next := ""
more := false
if end > len(all) {
end = len(all)
} else {
more = true
next = fmt.Sprintf("%d", end)
}
if off >= len(all) {
if start > len(allRows) {
return nil, "", false
}
return all[off:end], next, more
end := start + limit
next := ""
more := false
if end > len(allRows) {
end = len(allRows)
} else {
more = true
next = FormatPrefixPageCursor(int64(end - 1))
}
if start >= end {
return nil, "", false
}
out := make([]PrefixRow, end-start)
copy(out, allRows[start:end])
return out, next, more
}
func (m *Memory) ListGlobalSettings(tenantID string) (map[string]any, error) {
+30
View File
@@ -0,0 +1,30 @@
package store
import (
"strconv"
"strings"
)
// ParsePrefixPageCursor decodes opaque cursors for revision prefix pagination.
func ParsePrefixPageCursor(cursor string) (afterID *int64, offset int, useOffset bool) {
cursor = strings.TrimSpace(cursor)
if cursor == "" {
return nil, 0, false
}
if strings.HasPrefix(cursor, "o:") {
n, err := strconv.Atoi(strings.TrimPrefix(cursor, "o:"))
if err != nil || n < 0 {
return nil, 0, false
}
return nil, n, true
}
if n, err := strconv.ParseInt(cursor, 10, 64); err == nil && n >= 0 {
return &n, 0, false
}
return nil, 0, false
}
// FormatPrefixPageCursor encodes the keyset cursor (last row id or slice index).
func FormatPrefixPageCursor(lastID int64) string {
return strconv.FormatInt(lastID, 10)
}
@@ -0,0 +1,6 @@
DROP INDEX IF EXISTS idx_rev_mat_prefix_uniq;
DROP INDEX IF EXISTS idx_rev_mat_prefix_rev_cover;
CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_rev_id
ON revision_materialized_prefix (revision_id, id);
DROP INDEX IF EXISTS idx_module_tenant_active_sort;
DROP INDEX IF EXISTS idx_rev_mat_prefix_rev_prefix;
@@ -0,0 +1,17 @@
-- Query indexes: revision diff/list and module sort (H4 + M4).
CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_rev_prefix
ON revision_materialized_prefix (revision_id, prefix);
CREATE INDEX IF NOT EXISTS idx_module_tenant_active_sort
ON module (tenant_id, priority, name)
WHERE deleted_at IS NULL;
-- Covering index for prefix list pagination (PostgreSQL 11+ INCLUDE).
DROP INDEX IF EXISTS idx_rev_mat_prefix_rev_id;
CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_rev_cover
ON revision_materialized_prefix (revision_id, id)
INCLUDE (prefix, community_id, source);
CREATE UNIQUE INDEX IF NOT EXISTS idx_rev_mat_prefix_uniq
ON revision_materialized_prefix (revision_id, prefix, community_id, source);
@@ -0,0 +1,6 @@
UPDATE config_revision cr
SET meta_json = cr.meta_json || jsonb_build_object('preview_fragments', COALESCE(p.fragments, '{}'::jsonb))
FROM config_revision_preview p
WHERE p.revision_id = cr.id;
DROP TABLE IF EXISTS config_revision_preview;
@@ -0,0 +1,15 @@
-- Split BIRD preview fragments out of config_revision.meta_json (H1).
CREATE TABLE config_revision_preview (
revision_id UUID PRIMARY KEY REFERENCES config_revision (id) ON DELETE CASCADE,
fragments JSONB NOT NULL DEFAULT '{}'
);
INSERT INTO config_revision_preview (revision_id, fragments)
SELECT id, COALESCE(meta_json->'preview_fragments', '{}'::jsonb)
FROM config_revision
WHERE meta_json ? 'preview_fragments';
UPDATE config_revision
SET meta_json = meta_json - 'preview_fragments'
WHERE meta_json ? 'preview_fragments';
@@ -0,0 +1,3 @@
ALTER TABLE config_revision DROP COLUMN IF EXISTS prefix_snapshot_id;
DROP TABLE IF EXISTS prefix_snapshot_row;
DROP TABLE IF EXISTS prefix_snapshot;
@@ -0,0 +1,22 @@
-- Content-addressed prefix snapshots (H2 expand).
CREATE TABLE prefix_snapshot (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
content_hash CHAR(64) NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
CONSTRAINT prefix_snapshot_hash_uniq UNIQUE (content_hash)
);
CREATE TABLE prefix_snapshot_row (
snapshot_id UUID NOT NULL REFERENCES prefix_snapshot (id) ON DELETE CASCADE,
ord INTEGER NOT NULL,
prefix TEXT NOT NULL,
community_id UUID REFERENCES bgp_community (id) ON DELETE SET NULL,
source TEXT NOT NULL DEFAULT '',
PRIMARY KEY (snapshot_id, ord)
);
CREATE INDEX idx_prefix_snapshot_row_snapshot ON prefix_snapshot_row (snapshot_id);
ALTER TABLE config_revision
ADD COLUMN prefix_snapshot_id UUID REFERENCES prefix_snapshot (id) ON DELETE RESTRICT;
@@ -0,0 +1,3 @@
DELETE FROM prefix_snapshot_row;
UPDATE config_revision SET prefix_snapshot_id = NULL WHERE prefix_snapshot_id IS NOT NULL;
DELETE FROM prefix_snapshot;
@@ -0,0 +1,35 @@
-- Backfill prefix snapshots from revision_materialized_prefix.
-- prefix_snapshot_row.prefix is TEXT (revision_materialized_prefix.prefix since 000003).
ALTER TABLE prefix_snapshot_row
ALTER COLUMN prefix TYPE TEXT USING prefix::text;
WITH new_snaps AS (
INSERT INTO prefix_snapshot (id, content_hash)
SELECT gen_random_uuid(),
substr(replace(cr.id::text, '-', '') || replace(cr.id::text, '-', ''), 1, 64)
FROM config_revision cr
WHERE cr.prefix_snapshot_id IS NULL
AND EXISTS (
SELECT 1 FROM revision_materialized_prefix rmp WHERE rmp.revision_id = cr.id
)
RETURNING id, content_hash
)
UPDATE config_revision cr
SET prefix_snapshot_id = ns.id
FROM new_snaps ns
WHERE cr.prefix_snapshot_id IS NULL
AND ns.content_hash = substr(replace(cr.id::text, '-', '') || replace(cr.id::text, '-', ''), 1, 64);
INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source)
SELECT cr.prefix_snapshot_id,
(row_number() OVER (PARTITION BY cr.id ORDER BY rmp.id) - 1)::int,
rmp.prefix,
rmp.community_id,
rmp.source
FROM config_revision cr
JOIN revision_materialized_prefix rmp ON rmp.revision_id = cr.id
WHERE cr.prefix_snapshot_id IS NOT NULL
AND NOT EXISTS (
SELECT 1 FROM prefix_snapshot_row psr WHERE psr.snapshot_id = cr.prefix_snapshot_id
);
@@ -0,0 +1 @@
COMMENT ON COLUMN config_revision.prefix_snapshot_id IS NULL;
@@ -0,0 +1,3 @@
-- H2 contract marker: new revisions use prefix_snapshot_id only (enforced in application code).
COMMENT ON COLUMN config_revision.prefix_snapshot_id IS 'Materialized prefixes; revision_materialized_prefix deprecated for new rows';
@@ -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;
@@ -0,0 +1,46 @@
CREATE TABLE module_prefix_snapshot_row (
tenant_id UUID NOT NULL,
module_id UUID NOT NULL,
ord INTEGER NOT NULL,
prefix TEXT 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
);
-- prefixes_json from Go json.Marshal(PrefixRow) used "Prefix"/"CommunityID"/"Source" before json tags.
INSERT INTO module_prefix_snapshot_row (tenant_id, module_id, ord, prefix, community_id, source)
SELECT tenant_id,
module_id,
(row_number() OVER (PARTITION BY tenant_id, module_id ORDER BY ordinality) - 1)::int,
prefix,
NULLIF(community_id, '')::uuid,
COALESCE(source, '')
FROM (
SELECT mps.tenant_id,
mps.module_id,
t.ordinality,
COALESCE(
NULLIF(trim(t.elem->>'prefix'), ''),
NULLIF(trim(t.elem->>'Prefix'), '')
) AS prefix,
COALESCE(
NULLIF(trim(t.elem->>'community_id'), ''),
NULLIF(trim(t.elem->>'CommunityID'), '')
) AS community_id,
COALESCE(
NULLIF(trim(t.elem->>'source'), ''),
NULLIF(trim(t.elem->>'Source'), ''),
''
) AS 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
) parsed
WHERE parsed.prefix IS NOT NULL
AND trim(parsed.prefix) <> '';
ALTER TABLE module_prefix_snapshot DROP COLUMN prefixes_json;
@@ -0,0 +1,10 @@
ALTER TABLE asn_prefix_cache ADD COLUMN prefixes_json JSONB NOT NULL DEFAULT '[]';
UPDATE asn_prefix_cache apc
SET prefixes_json = COALESCE((
SELECT jsonb_agg(apcr.prefix::text ORDER BY apcr.prefix::text)
FROM asn_prefix_cache_row apcr
WHERE apcr.asn = apc.asn
), '[]'::jsonb);
DROP TABLE IF EXISTS asn_prefix_cache_row;
@@ -0,0 +1,14 @@
CREATE TABLE asn_prefix_cache_row (
asn BIGINT NOT NULL REFERENCES asn_prefix_cache (asn) ON DELETE CASCADE,
prefix CIDR NOT NULL,
PRIMARY KEY (asn, prefix)
);
INSERT INTO asn_prefix_cache_row (asn, prefix)
SELECT apc.asn, t.elem::cidr
FROM asn_prefix_cache apc
CROSS JOIN LATERAL jsonb_array_elements_text(apc.prefixes_json) AS t(elem)
WHERE jsonb_typeof(apc.prefixes_json) = 'array'
AND jsonb_array_length(apc.prefixes_json) > 0;
ALTER TABLE asn_prefix_cache DROP COLUMN prefixes_json;
@@ -0,0 +1 @@
DROP INDEX IF EXISTS idx_job_audit_created_brin;
@@ -0,0 +1,2 @@
CREATE INDEX IF NOT EXISTS idx_job_audit_created_brin
ON job_audit USING BRIN (created_at);
@@ -0,0 +1,18 @@
CREATE TABLE IF NOT EXISTS module_cdn_fetch_log (
id BIGSERIAL PRIMARY KEY,
module_id UUID NOT NULL REFERENCES module (id) ON DELETE CASCADE,
source_id UUID REFERENCES module_cdn_source (id) ON DELETE SET NULL,
http_status INTEGER,
bytes BIGINT,
error TEXT,
fetched_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_module_cdn_fetch_log_module ON module_cdn_fetch_log (module_id, fetched_at DESC);
ALTER TABLE revision_materialized_prefix ADD COLUMN IF NOT EXISTS meta_json JSONB NOT NULL DEFAULT '{}';
ALTER TABLE module_domain_entry ADD COLUMN IF NOT EXISTS resolve_meta JSONB NOT NULL DEFAULT '{}';
CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_revision ON revision_materialized_prefix (revision_id);
CREATE INDEX IF NOT EXISTS idx_module_tenant ON module (tenant_id);
CREATE INDEX IF NOT EXISTS idx_module_prefix_snapshot_collected ON module_prefix_snapshot (collected_at);
CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_value ON revision_materialized_prefix (prefix);
@@ -0,0 +1,8 @@
ALTER TABLE revision_materialized_prefix DROP COLUMN IF EXISTS meta_json;
ALTER TABLE module_domain_entry DROP COLUMN IF EXISTS resolve_meta;
DROP TABLE IF EXISTS module_cdn_fetch_log;
DROP INDEX IF EXISTS idx_rev_mat_prefix_revision;
DROP INDEX IF EXISTS idx_module_tenant;
DROP INDEX IF EXISTS idx_module_prefix_snapshot_collected;
DROP INDEX IF EXISTS idx_rev_mat_prefix_value;
@@ -0,0 +1,3 @@
DROP INDEX IF EXISTS idx_rev_mat_prefix_uniq;
DROP INDEX IF EXISTS idx_module_tenant_active_sort;
DROP INDEX IF EXISTS idx_rev_mat_prefix_rev_prefix;
@@ -0,0 +1,9 @@
CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_rev_prefix
ON revision_materialized_prefix (revision_id, prefix);
CREATE INDEX IF NOT EXISTS idx_module_tenant_active_sort
ON module (tenant_id, priority, name)
WHERE deleted_at IS NULL;
CREATE UNIQUE INDEX IF NOT EXISTS idx_rev_mat_prefix_uniq
ON revision_materialized_prefix (revision_id, prefix, community_id, source);
@@ -0,0 +1,9 @@
UPDATE config_revision
SET meta_json = json_set(
meta_json,
'$.preview_fragments',
json(COALESCE((SELECT fragments FROM config_revision_preview p WHERE p.revision_id = config_revision.id), '{}'))
)
WHERE id IN (SELECT revision_id FROM config_revision_preview);
DROP TABLE IF EXISTS config_revision_preview;
@@ -0,0 +1,13 @@
CREATE TABLE config_revision_preview (
revision_id TEXT NOT NULL PRIMARY KEY REFERENCES config_revision (id) ON DELETE CASCADE,
fragments TEXT NOT NULL DEFAULT '{}'
);
INSERT INTO config_revision_preview (revision_id, fragments)
SELECT id, json(COALESCE(json_extract(meta_json, '$.preview_fragments'), '{}'))
FROM config_revision
WHERE json_extract(meta_json, '$.preview_fragments') IS NOT NULL;
UPDATE config_revision
SET meta_json = json_remove(meta_json, '$.preview_fragments')
WHERE json_extract(meta_json, '$.preview_fragments') IS NOT NULL;
@@ -0,0 +1,3 @@
ALTER TABLE config_revision DROP COLUMN prefix_snapshot_id;
DROP TABLE IF EXISTS prefix_snapshot_row;
DROP TABLE IF EXISTS prefix_snapshot;
@@ -0,0 +1,18 @@
CREATE TABLE prefix_snapshot (
id TEXT NOT NULL PRIMARY KEY,
content_hash TEXT NOT NULL UNIQUE,
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);
CREATE TABLE prefix_snapshot_row (
snapshot_id TEXT NOT NULL REFERENCES prefix_snapshot (id) ON DELETE CASCADE,
ord INTEGER NOT NULL,
prefix TEXT NOT NULL,
community_id TEXT,
source TEXT NOT NULL DEFAULT '',
PRIMARY KEY (snapshot_id, ord)
);
CREATE INDEX idx_prefix_snapshot_row_snapshot ON prefix_snapshot_row (snapshot_id);
ALTER TABLE config_revision ADD COLUMN prefix_snapshot_id TEXT REFERENCES prefix_snapshot (id) ON DELETE RESTRICT;
@@ -0,0 +1,3 @@
DELETE FROM prefix_snapshot_row;
UPDATE config_revision SET prefix_snapshot_id = NULL;
DELETE FROM prefix_snapshot;
@@ -0,0 +1,30 @@
-- SQLite backfill: one snapshot per revision with materialized prefixes.
INSERT INTO prefix_snapshot (id, content_hash)
SELECT lower(hex(randomblob(16))),
substr(replace(cr.id, '-', '') || replace(cr.id, '-', ''), 1, 64)
FROM config_revision cr
WHERE cr.prefix_snapshot_id IS NULL
AND EXISTS (SELECT 1 FROM revision_materialized_prefix rmp WHERE rmp.revision_id = cr.id);
UPDATE config_revision
SET prefix_snapshot_id = (
SELECT ps.id FROM prefix_snapshot ps
WHERE ps.content_hash = substr(replace(config_revision.id, '-', '') || replace(config_revision.id, '-', ''), 1, 64)
)
WHERE prefix_snapshot_id IS NULL
AND EXISTS (SELECT 1 FROM revision_materialized_prefix rmp WHERE rmp.revision_id = config_revision.id);
INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source)
SELECT cr.prefix_snapshot_id,
(SELECT COUNT(*) FROM revision_materialized_prefix r2
WHERE r2.revision_id = cr.id AND r2.id <= rmp.id) - 1,
rmp.prefix,
rmp.community_id,
rmp.source
FROM config_revision cr
JOIN revision_materialized_prefix rmp ON rmp.revision_id = cr.id
WHERE cr.prefix_snapshot_id IS NOT NULL
AND NOT EXISTS (
SELECT 1 FROM prefix_snapshot_row psr WHERE psr.snapshot_id = cr.prefix_snapshot_id
);
@@ -0,0 +1 @@
-- H2 contract marker (application stops writing revision_materialized_prefix for new revisions).
@@ -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;
@@ -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;
@@ -0,0 +1,2 @@
ALTER TABLE asn_prefix_cache ADD COLUMN prefixes_json TEXT NOT NULL DEFAULT '[]';
DROP TABLE IF EXISTS asn_prefix_cache_row;
@@ -0,0 +1,7 @@
CREATE TABLE asn_prefix_cache_row (
asn INTEGER NOT NULL REFERENCES asn_prefix_cache (asn) ON DELETE CASCADE,
prefix TEXT NOT NULL,
PRIMARY KEY (asn, prefix)
);
ALTER TABLE asn_prefix_cache DROP COLUMN prefixes_json;
@@ -0,0 +1 @@
DROP INDEX IF EXISTS idx_job_audit_created_brin;
@@ -0,0 +1 @@
CREATE INDEX IF NOT EXISTS idx_job_audit_created_brin ON job_audit (created_at);
@@ -0,0 +1,18 @@
ALTER TABLE revision_materialized_prefix ADD COLUMN meta_json TEXT NOT NULL DEFAULT '{}';
ALTER TABLE module_domain_entry ADD COLUMN resolve_meta TEXT NOT NULL DEFAULT '{}';
CREATE TABLE IF NOT EXISTS module_cdn_fetch_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
module_id TEXT NOT NULL REFERENCES module (id) ON DELETE CASCADE,
source_id TEXT REFERENCES module_cdn_source (id) ON DELETE SET NULL,
http_status INTEGER,
bytes INTEGER,
error TEXT,
fetched_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);
CREATE INDEX IF NOT EXISTS idx_module_cdn_fetch_log_module ON module_cdn_fetch_log (module_id, fetched_at DESC);
CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_revision ON revision_materialized_prefix (revision_id);
CREATE INDEX IF NOT EXISTS idx_module_tenant ON module (tenant_id);
CREATE INDEX IF NOT EXISTS idx_module_prefix_snapshot_collected ON module_prefix_snapshot (collected_at);
CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_value ON revision_materialized_prefix (prefix);
@@ -0,0 +1,8 @@
ALTER TABLE revision_materialized_prefix DROP COLUMN meta_json;
ALTER TABLE module_domain_entry DROP COLUMN resolve_meta;
DROP TABLE IF EXISTS module_cdn_fetch_log;
DROP INDEX IF EXISTS idx_rev_mat_prefix_revision;
DROP INDEX IF EXISTS idx_module_tenant;
DROP INDEX IF EXISTS idx_module_prefix_snapshot_collected;
DROP INDEX IF EXISTS idx_rev_mat_prefix_value;