Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
930e42b0b0 | ||
|
|
16b4923bd7 | ||
|
|
1cfd062835 | ||
|
|
21233bd578 | ||
|
|
990cc739df | ||
|
|
3500bd4624 | ||
|
|
374575ec01 | ||
|
|
f57b430052 | ||
|
|
8a9d60a5a7 | ||
|
|
b963311b43 | ||
|
|
50bdb8232b | ||
|
|
ee8e24ffc6 | ||
|
|
44b94caacf |
@@ -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 на больших таблицах.
|
||||||
@@ -17,6 +17,8 @@ type Deps struct {
|
|||||||
Store store.Backend
|
Store store.Backend
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var lastMaintenance time.Time
|
||||||
|
|
||||||
// Run blocks until ctx is cancelled.
|
// Run blocks until ctx is cancelled.
|
||||||
func Run(ctx context.Context, deps *Deps) {
|
func Run(ctx context.Context, deps *Deps) {
|
||||||
cfg := config.Load()
|
cfg := config.Load()
|
||||||
@@ -34,6 +36,10 @@ func Run(ctx context.Context, deps *Deps) {
|
|||||||
log.Printf("evobgp-ingest: stopped")
|
log.Printf("evobgp-ingest: stopped")
|
||||||
return
|
return
|
||||||
case <-t.C:
|
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)
|
prefetchCtx, cancel := context.WithTimeout(ctx, 50*time.Second)
|
||||||
err := pipeline.PrefetchCDNSourceETags(prefetchCtx, deps.Store, hc)
|
err := pipeline.PrefetchCDNSourceETags(prefetchCtx, deps.Store, hc)
|
||||||
cancel()
|
cancel()
|
||||||
|
|||||||
@@ -11,14 +11,22 @@ import (
|
|||||||
"github.com/jackc/pgx/v5"
|
"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) {
|
func (p *Postgres) GetASNPrefixCache(asn int64) (*store.ASNPrefixCacheEntry, bool, error) {
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
var holder string
|
var holder string
|
||||||
var fetchedAt time.Time
|
var fetchedAt time.Time
|
||||||
var raw []byte
|
|
||||||
err := p.pool.QueryRow(ctx, `
|
err := p.pool.QueryRow(ctx, `
|
||||||
SELECT holder, fetched_at, prefixes_json FROM asn_prefix_cache WHERE asn = $1`, asn).
|
SELECT holder, fetched_at FROM asn_prefix_cache WHERE asn = $1`, asn).
|
||||||
Scan(&holder, &fetchedAt, &raw)
|
Scan(&holder, &fetchedAt)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if errors.Is(err, pgx.ErrNoRows) {
|
if errors.Is(err, pgx.ErrNoRows) {
|
||||||
return nil, false, nil
|
return nil, false, nil
|
||||||
@@ -26,8 +34,25 @@ func (p *Postgres) GetASNPrefixCache(asn int64) (*store.ASNPrefixCacheEntry, boo
|
|||||||
return nil, false, err
|
return nil, false, err
|
||||||
}
|
}
|
||||||
var prefixes []string
|
var prefixes []string
|
||||||
if len(raw) > 0 {
|
if asnCacheRowTableExists(ctx, p.pool) {
|
||||||
_ = json.Unmarshal(raw, &prefixes)
|
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{
|
return &store.ASNPrefixCacheEntry{
|
||||||
ASN: asn,
|
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 {
|
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 {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
ctx := context.Background()
|
defer func() { _ = tx.Rollback(ctx) }()
|
||||||
_, err = p.pool.Exec(ctx, `
|
_, err = tx.Exec(ctx, `
|
||||||
INSERT INTO asn_prefix_cache (asn, holder, prefixes_json, fetched_at)
|
INSERT INTO asn_prefix_cache (asn, holder, fetched_at)
|
||||||
VALUES ($1, $2, $3::jsonb, now())
|
VALUES ($1, $2, now())
|
||||||
ON CONFLICT (asn) DO UPDATE SET
|
ON CONFLICT (asn) DO UPDATE SET
|
||||||
holder = EXCLUDED.holder,
|
holder = EXCLUDED.holder,
|
||||||
prefixes_json = EXCLUDED.prefixes_json,
|
fetched_at = EXCLUDED.fetched_at`, asn, holder)
|
||||||
fetched_at = EXCLUDED.fetched_at`,
|
if err != nil {
|
||||||
asn, holder, string(raw))
|
return err
|
||||||
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)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
@@ -12,16 +12,24 @@ import (
|
|||||||
"github.com/jackc/pgx/v5"
|
"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) {
|
func (p *Postgres) GetModulePrefixSnapshot(tenantID, moduleID string) (*store.ModulePrefixSnapshot, bool, error) {
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
var inputHash string
|
var inputHash string
|
||||||
var collectedAt time.Time
|
var collectedAt time.Time
|
||||||
var raw []byte
|
|
||||||
err := p.pool.QueryRow(ctx, `
|
err := p.pool.QueryRow(ctx, `
|
||||||
SELECT input_hash, collected_at, prefixes_json
|
SELECT input_hash, collected_at
|
||||||
FROM module_prefix_snapshot
|
FROM module_prefix_snapshot
|
||||||
WHERE tenant_id = $1 AND module_id = $2`,
|
WHERE tenant_id = $1 AND module_id = $2`,
|
||||||
tenantID, moduleID).Scan(&inputHash, &collectedAt, &raw)
|
tenantID, moduleID).Scan(&inputHash, &collectedAt)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if errors.Is(err, pgx.ErrNoRows) {
|
if errors.Is(err, pgx.ErrNoRows) {
|
||||||
return nil, false, nil
|
return nil, false, nil
|
||||||
@@ -29,9 +37,31 @@ func (p *Postgres) GetModulePrefixSnapshot(tenantID, moduleID string) (*store.Mo
|
|||||||
return nil, false, err
|
return nil, false, err
|
||||||
}
|
}
|
||||||
var prefixes []store.PrefixRow
|
var prefixes []store.PrefixRow
|
||||||
if len(raw) > 0 {
|
if moduleSnapshotRowTableExists(ctx, p.pool) {
|
||||||
if err := json.Unmarshal(raw, &prefixes); err != nil {
|
rows, qerr := p.pool.Query(ctx, `
|
||||||
return nil, false, err
|
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{
|
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) == "" {
|
if strings.TrimSpace(tenantID) == "" || strings.TrimSpace(moduleID) == "" || strings.TrimSpace(inputHash) == "" {
|
||||||
return store.ErrInvalidInput
|
return store.ErrInvalidInput
|
||||||
}
|
}
|
||||||
raw, err := json.Marshal(prefixes)
|
ctx := context.Background()
|
||||||
|
tx, err := p.pool.Begin(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
ctx := context.Background()
|
defer func() { _ = tx.Rollback(ctx) }()
|
||||||
_, err = p.pool.Exec(ctx, `
|
_, err = tx.Exec(ctx, `
|
||||||
INSERT INTO module_prefix_snapshot (tenant_id, module_id, input_hash, collected_at, prefixes_json)
|
INSERT INTO module_prefix_snapshot (tenant_id, module_id, input_hash, collected_at)
|
||||||
VALUES ($1::uuid, $2::uuid, $3, now(), $4::jsonb)
|
VALUES ($1::uuid, $2::uuid, $3, now())
|
||||||
ON CONFLICT (tenant_id, module_id) DO UPDATE SET
|
ON CONFLICT (tenant_id, module_id) DO UPDATE SET
|
||||||
input_hash = EXCLUDED.input_hash,
|
input_hash = EXCLUDED.input_hash,
|
||||||
collected_at = EXCLUDED.collected_at,
|
collected_at = EXCLUDED.collected_at`,
|
||||||
prefixes_json = EXCLUDED.prefixes_json`,
|
tenantID, moduleID, inputHash)
|
||||||
tenantID, moduleID, inputHash, string(raw))
|
if err != nil {
|
||||||
return err
|
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 {
|
func (p *Postgres) DeleteModulePrefixSnapshot(tenantID, moduleID string) error {
|
||||||
|
|||||||
@@ -687,7 +687,7 @@ func (p *Postgres) GetRevision(tenantID, revisionID string) (*store.Revision, er
|
|||||||
if mj.PreviewFragments == nil {
|
if mj.PreviewFragments == nil {
|
||||||
mj.PreviewFragments = map[string]string{}
|
mj.PreviewFragments = map[string]string{}
|
||||||
}
|
}
|
||||||
r.PreviewFragments = mj.PreviewFragments
|
r.PreviewFragments = loadRevisionPreview(ctx, p.pool, revisionID, mj.PreviewFragments)
|
||||||
r.MaterializedPrefixCount = mj.MaterializedPrefixCount
|
r.MaterializedPrefixCount = mj.MaterializedPrefixCount
|
||||||
return &r, nil
|
return &r, nil
|
||||||
}
|
}
|
||||||
@@ -789,12 +789,7 @@ func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor stri
|
|||||||
if limit <= 0 {
|
if limit <= 0 {
|
||||||
limit = 50
|
limit = 50
|
||||||
}
|
}
|
||||||
off := 0
|
afterID, off, useOffset := store.ParsePrefixPageCursor(cursor)
|
||||||
if cursor != "" {
|
|
||||||
if n, err := strconv.Atoi(cursor); err == nil && n >= 0 {
|
|
||||||
off = n
|
|
||||||
}
|
|
||||||
}
|
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
var one int
|
var one int
|
||||||
if err := p.pool.QueryRow(ctx, `
|
if err := p.pool.QueryRow(ctx, `
|
||||||
@@ -805,34 +800,55 @@ func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor stri
|
|||||||
}
|
}
|
||||||
return nil, "", false
|
return nil, "", false
|
||||||
}
|
}
|
||||||
rows, err := p.pool.Query(ctx, `
|
if snapID, ok := p.revisionPrefixSnapshotID(ctx, revisionID); ok {
|
||||||
SELECT prefix::text, community_id::text, source FROM revision_materialized_prefix
|
return p.listSnapshotPrefixes(ctx, snapID, cursor, limit)
|
||||||
WHERE revision_id=$1 ORDER BY id
|
}
|
||||||
LIMIT $2 OFFSET $3`, revisionID, limit+1, off)
|
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 {
|
if err != nil {
|
||||||
return nil, "", false
|
return nil, "", false
|
||||||
}
|
}
|
||||||
defer rows.Close()
|
defer rows.Close()
|
||||||
var all []store.PrefixRow
|
var all []store.PrefixRow
|
||||||
|
var ids []int64
|
||||||
for rows.Next() {
|
for rows.Next() {
|
||||||
|
var rowID int64
|
||||||
var pr store.PrefixRow
|
var pr store.PrefixRow
|
||||||
var comm *string
|
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
|
continue
|
||||||
}
|
}
|
||||||
pr.CommunityID = comm
|
pr.CommunityID = comm
|
||||||
|
ids = append(ids, rowID)
|
||||||
all = append(all, pr)
|
all = append(all, pr)
|
||||||
}
|
}
|
||||||
agentDebugNDJSON3214("B", "repository/postgres.go:ListRevisionPrefixes", "list_prefixes_fetched", map[string]any{
|
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
|
more := len(all) > limit
|
||||||
if more {
|
if more {
|
||||||
all = all[:limit]
|
all = all[:limit]
|
||||||
|
ids = ids[:limit]
|
||||||
}
|
}
|
||||||
next := ""
|
next := ""
|
||||||
if more {
|
if more && len(ids) > 0 {
|
||||||
next = fmt.Sprintf("%d", off+limit)
|
next = store.FormatPrefixPageCursor(ids[len(ids)-1])
|
||||||
}
|
}
|
||||||
if len(all) == 0 {
|
if len(all) == 0 {
|
||||||
return nil, "", false
|
return nil, "", false
|
||||||
@@ -848,26 +864,35 @@ func (p *Postgres) CreateRollbackRevision(tenantID, sourceRevisionID string) (st
|
|||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
newID := uuid.NewString()
|
newID := uuid.NewString()
|
||||||
parent := sourceRevisionID
|
parent := sourceRevisionID
|
||||||
meta, _ := json.Marshal(map[string]any{
|
meta, err := revisionMetaWithoutPreview(src.MaterializedPrefixCount)
|
||||||
"preview_fragments": src.PreviewFragments,
|
if err != nil {
|
||||||
"materialized_prefix_count": src.MaterializedPrefixCount,
|
return "", err
|
||||||
})
|
}
|
||||||
var modArg any
|
var modArg any
|
||||||
if strings.TrimSpace(src.ModuleID) != "" {
|
if strings.TrimSpace(src.ModuleID) != "" {
|
||||||
modArg = src.ModuleID
|
modArg = src.ModuleID
|
||||||
}
|
}
|
||||||
_, err = p.pool.Exec(ctx, `
|
tx, err := p.pool.Begin(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))
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
// copy materialized prefixes
|
defer func() { _ = tx.Rollback(ctx) }()
|
||||||
_, _ = p.pool.Exec(ctx, `
|
_, err = tx.Exec(ctx, `
|
||||||
INSERT INTO revision_materialized_prefix (revision_id, prefix, community_id, source, meta_json)
|
INSERT INTO config_revision (id, tenant_id, module_id, content_hash, parent_revision_id, meta_json)
|
||||||
SELECT $1::uuid, prefix, community_id, source, meta_json FROM revision_materialized_prefix WHERE revision_id=$2::uuid`,
|
VALUES ($1,$2,$3,$4,$5::uuid,$6::jsonb)`,
|
||||||
newID, sourceRevisionID)
|
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
|
return newID, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -884,20 +909,18 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
|
|||||||
var unchanged int
|
var unchanged int
|
||||||
err := p.pool.QueryRow(ctx, `
|
err := p.pool.QueryRow(ctx, `
|
||||||
SELECT COUNT(*)::int FROM (
|
SELECT COUNT(*)::int FROM (
|
||||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
|
SELECT b.prefix FROM (`+sqlRevisionPrefixes("$2")+`) b
|
||||||
INTERSECT
|
INNER JOIN (`+sqlRevisionPrefixes("$1")+`) a ON a.prefix = b.prefix
|
||||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
|
|
||||||
) t`, aID, bID).Scan(&unchanged)
|
) t`, aID, bID).Scan(&unchanged)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
// added: в B, нет в A; removed: в A, нет в B — без загрузки полных снапшотов в память.
|
|
||||||
rowsAdded, err := p.pool.Query(ctx, `
|
rowsAdded, err := p.pool.Query(ctx, `
|
||||||
SELECT prefix::text FROM (
|
SELECT b.prefix::text FROM (`+sqlRevisionPrefixes("$2")+`) b
|
||||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
|
LEFT JOIN (`+sqlRevisionPrefixes("$1")+`) a ON a.prefix = b.prefix
|
||||||
EXCEPT
|
WHERE a.prefix IS NULL
|
||||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
|
ORDER BY b.prefix
|
||||||
) s ORDER BY 1 LIMIT $3`, bID, aID, maxRevisionDiffRows+1)
|
LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -916,11 +939,11 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
|
|||||||
}
|
}
|
||||||
addedTruncated := len(added) >= maxRevisionDiffRows
|
addedTruncated := len(added) >= maxRevisionDiffRows
|
||||||
rowsRem, err := p.pool.Query(ctx, `
|
rowsRem, err := p.pool.Query(ctx, `
|
||||||
SELECT prefix::text FROM (
|
SELECT a.prefix::text FROM (`+sqlRevisionPrefixes("$1")+`) a
|
||||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
|
LEFT JOIN (`+sqlRevisionPrefixes("$2")+`) b ON b.prefix = a.prefix
|
||||||
EXCEPT
|
WHERE b.prefix IS NULL
|
||||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
|
ORDER BY a.prefix
|
||||||
) s ORDER BY 1 LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
|
LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -1056,10 +1079,7 @@ func (p *Postgres) CreateRenderRevision(revisionID, tenantID, moduleID string, p
|
|||||||
if previewFragments == nil {
|
if previewFragments == nil {
|
||||||
previewFragments = map[string]string{}
|
previewFragments = map[string]string{}
|
||||||
}
|
}
|
||||||
meta, err := json.Marshal(map[string]any{
|
meta, err := revisionMetaWithoutPreview(len(prefixes))
|
||||||
"preview_fragments": previewFragments,
|
|
||||||
"materialized_prefix_count": len(prefixes),
|
|
||||||
})
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -1073,32 +1093,23 @@ func (p *Postgres) CreateRenderRevision(revisionID, tenantID, moduleID string, p
|
|||||||
if parentRevisionID != nil && strings.TrimSpace(*parentRevisionID) != "" {
|
if parentRevisionID != nil && strings.TrimSpace(*parentRevisionID) != "" {
|
||||||
parent = strings.TrimSpace(*parentRevisionID)
|
parent = strings.TrimSpace(*parentRevisionID)
|
||||||
}
|
}
|
||||||
|
revID := strings.TrimSpace(revisionID)
|
||||||
_, err = tx.Exec(ctx, `
|
_, err = tx.Exec(ctx, `
|
||||||
INSERT INTO config_revision (id, tenant_id, module_id, content_hash, parent_revision_id, meta_json)
|
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)`,
|
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 {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if len(prefixes) > 0 {
|
if err := insertRevisionPreview(ctx, tx, revID, previewFragments); err != nil {
|
||||||
_, err = tx.CopyFrom(ctx,
|
return err
|
||||||
pgx.Identifier{"revision_materialized_prefix"},
|
}
|
||||||
[]string{"revision_id", "prefix", "community_id", "source"},
|
snapID, err := p.ensurePrefixSnapshot(ctx, tx, contentHash, prefixes)
|
||||||
pgx.CopyFromSlice(len(prefixes), func(i int) ([]any, error) {
|
if err != nil {
|
||||||
pr := prefixes[i]
|
return err
|
||||||
var comm any
|
}
|
||||||
if pr.CommunityID != nil && strings.TrimSpace(*pr.CommunityID) != "" {
|
if err := p.linkRevisionPrefixSnapshot(ctx, tx, revID, snapID); err != nil {
|
||||||
comm = strings.TrimSpace(*pr.CommunityID)
|
return err
|
||||||
}
|
|
||||||
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 := tx.Commit(ctx); err != nil {
|
if err := tx.Commit(ctx); err != nil {
|
||||||
return err
|
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
|
||||||
|
}
|
||||||
@@ -26,9 +26,8 @@ func (p *Postgres) seedDemo(ctx context.Context) error {
|
|||||||
p1 := uuid.NewString()
|
p1 := uuid.NewString()
|
||||||
p2 := uuid.NewString()
|
p2 := uuid.NewString()
|
||||||
|
|
||||||
preview := map[string]any{
|
previewFrags := map[string]string{
|
||||||
"preview_fragments": map[string]string{
|
"bird.conf": `# EvoBGP demo bundle
|
||||||
"bird.conf": `# EvoBGP demo bundle
|
|
||||||
router id 192.0.2.1;
|
router id 192.0.2.1;
|
||||||
|
|
||||||
protocol device {
|
protocol device {
|
||||||
@@ -39,15 +38,12 @@ protocol direct {
|
|||||||
ipv6;
|
ipv6;
|
||||||
}
|
}
|
||||||
`,
|
`,
|
||||||
"bird.d/evobgp_demo.conf": "# static demo fragment\n",
|
"bird.d/evobgp_demo.conf": "# static demo fragment\n",
|
||||||
},
|
|
||||||
"materialized_prefix_count": 128,
|
|
||||||
}
|
}
|
||||||
previewB, _ := json.Marshal(preview)
|
previewMeta, _ := json.Marshal(map[string]any{"materialized_prefix_count": 128})
|
||||||
parentMeta, _ := json.Marshal(map[string]any{
|
parentMeta, _ := json.Marshal(map[string]any{"materialized_prefix_count": 0})
|
||||||
"preview_fragments": map[string]string{"bird.conf": "# parent revision\n"},
|
parentPreview, _ := json.Marshal(map[string]string{"bird.conf": "# parent revision\n"})
|
||||||
"materialized_prefix_count": 0,
|
previewFragsB, _ := json.Marshal(previewFrags)
|
||||||
})
|
|
||||||
|
|
||||||
tx, err := p.pool.Begin(ctx)
|
tx, err := p.pool.Begin(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -77,13 +73,32 @@ protocol direct {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if _, err := tx.Exec(ctx, `
|
if _, err := tx.Exec(ctx, `
|
||||||
INSERT INTO config_revision (id, tenant_id, content_hash, parent_revision_id, meta_json)
|
INSERT INTO config_revision_preview (revision_id, fragments) VALUES ($1::uuid, $2::jsonb)`, parent, string(parentPreview)); err != nil {
|
||||||
VALUES ($1,$2,'sha256:demo-rev-1',$3::uuid,$4::jsonb)`, rid, tid, parent, string(previewB)); err != nil {
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if _, err := tx.Exec(ctx, `
|
if _, err := tx.Exec(ctx, `
|
||||||
INSERT INTO revision_materialized_prefix (revision_id, prefix, community_id, source)
|
INSERT INTO config_revision (id, tenant_id, content_hash, parent_revision_id, meta_json)
|
||||||
VALUES ($1::uuid,'203.0.113.0/24',$2::uuid,'demo'), ($1::uuid,'2001:db8::/32',$2::uuid,'demo')`, rid, cid); err != nil {
|
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
|
return err
|
||||||
}
|
}
|
||||||
if _, err := tx.Exec(ctx, `
|
if _, err := tx.Exec(ctx, `
|
||||||
|
|||||||
@@ -112,6 +112,9 @@ type Backend interface {
|
|||||||
|
|
||||||
// Ping verifies backend connectivity (no-op for in-memory).
|
// Ping verifies backend connectivity (no-op for in-memory).
|
||||||
Ping(ctx context.Context) error
|
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.
|
// 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.
|
// PrefixRow is one materialized prefix for GET /revisions/.../prefixes.
|
||||||
type PrefixRow struct {
|
type PrefixRow struct {
|
||||||
Prefix string
|
Prefix string `json:"prefix"`
|
||||||
CommunityID *string
|
CommunityID *string `json:"community_id,omitempty"`
|
||||||
Source string
|
Source string `json:"source,omitempty"`
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -305,6 +305,11 @@ func (m *Memory) Ping(ctx context.Context) error {
|
|||||||
return nil
|
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.
|
// ListTenantIDs returns tenant ids sorted lexicographically.
|
||||||
func (m *Memory) ListTenantIDs() ([]string, error) {
|
func (m *Memory) ListTenantIDs() ([]string, error) {
|
||||||
m.mu.RLock()
|
m.mu.RLock()
|
||||||
|
|||||||
@@ -1,8 +1,6 @@
|
|||||||
package store
|
package store
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"fmt"
|
|
||||||
"strconv"
|
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -795,31 +793,37 @@ func (m *Memory) ListRevisionPrefixes(tenantID, revisionID string, cursor string
|
|||||||
if limit <= 0 {
|
if limit <= 0 {
|
||||||
limit = 50
|
limit = 50
|
||||||
}
|
}
|
||||||
|
afterID, off, useOffset := ParsePrefixPageCursor(cursor)
|
||||||
m.mu.RLock()
|
m.mu.RLock()
|
||||||
defer m.mu.RUnlock()
|
defer m.mu.RUnlock()
|
||||||
if _, err := m.getRevisionLocked(tenantID, revisionID); err != nil {
|
if _, err := m.getRevisionLocked(tenantID, revisionID); err != nil {
|
||||||
return nil, "", false
|
return nil, "", false
|
||||||
}
|
}
|
||||||
all := m.revPrefixes[revisionID]
|
allRows := m.revPrefixes[revisionID]
|
||||||
off := 0
|
start := 0
|
||||||
if cursor != "" {
|
if useOffset {
|
||||||
if n, err := strconv.Atoi(cursor); err == nil && n >= 0 {
|
start = off
|
||||||
off = n
|
} else if afterID != nil {
|
||||||
}
|
start = int(*afterID) + 1
|
||||||
}
|
}
|
||||||
end := off + limit
|
if start > len(allRows) {
|
||||||
next := ""
|
|
||||||
more := false
|
|
||||||
if end > len(all) {
|
|
||||||
end = len(all)
|
|
||||||
} else {
|
|
||||||
more = true
|
|
||||||
next = fmt.Sprintf("%d", end)
|
|
||||||
}
|
|
||||||
if off >= len(all) {
|
|
||||||
return nil, "", false
|
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) {
|
func (m *Memory) ListGlobalSettings(tenantID string) (map[string]any, error) {
|
||||||
|
|||||||
@@ -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;
|
||||||
Reference in New Issue
Block a user