refactor(db): normalize asn prefix cache and ttl cleanup

Строки asn_prefix_cache_row; периодический prune через evobgp-ingest.

Co-authored-by: Cursor <[email protected]>

internal/ingest/run.go

@@ -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()

internal/repository/asn_cache.go

@@ -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)
 }

internal/repository/maintenance.go

@@ -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)
+}

internal/store/backend.go

@@ -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.

internal/store/memory.go

@@ -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()

migrations/postgres/000020_asn_cache_rows.down.sql

@@ -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;

migrations/postgres/000020_asn_cache_rows.up.sql

@@ -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;

migrations/sqlite/000020_asn_cache_rows.down.sql

@@ -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;

migrations/sqlite/000020_asn_cache_rows.up.sql

@@ -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;