package repository import ( "context" "time" "evobgp/internal/store" ) const revisionPruneBatchSize = 50 // sqlPrunableRevisionsWhere appends prunable revision predicates (tenant + cutoff params). func sqlPrunableRevisionsWhere(tenantParam, cutoffParam string) string { return `tenant_id = ` + tenantParam + ` AND created_at < ` + cutoffParam + ` AND id <> ( SELECT id FROM config_revision WHERE tenant_id = ` + tenantParam + ` ORDER BY created_at DESC LIMIT 1 ) AND NOT EXISTS ( SELECT 1 FROM bgp_speaker AS sp WHERE sp.tenant_id = ` + tenantParam + ` AND (sp.last_applied_revision_id = config_revision.id OR sp.published_revision_id = config_revision.id) )` } func (p *Postgres) EstimateRevisionPrune(tenantID string, cutoff time.Time, retentionMinutes int) (store.RevisionPruneEstimate, error) { ctx := context.Background() out := store.RevisionPruneEstimate{ RetentionMinutes: retentionMinutes, CutoffAt: cutoff.UTC(), } where := sqlPrunableRevisionsWhere("$1", "$2") var revBytes, previewBytes, rmpBytes, snapRowBytes int64 err := p.pool.QueryRow(ctx, ` WITH prunable AS ( SELECT id, prefix_snapshot_id, content_hash, meta_json FROM config_revision WHERE `+where+` ), freed_snaps AS ( SELECT DISTINCT pr.prefix_snapshot_id AS snap_id FROM prunable pr WHERE pr.prefix_snapshot_id IS NOT NULL EXCEPT SELECT DISTINCT cr.prefix_snapshot_id FROM config_revision cr WHERE cr.prefix_snapshot_id IS NOT NULL AND cr.id NOT IN (SELECT id FROM prunable) ) SELECT (SELECT COUNT(*)::int FROM prunable), COALESCE((SELECT SUM(octet_length(content_hash) + octet_length(meta_json::text))::bigint FROM prunable), 0), COALESCE(( SELECT SUM(octet_length(value))::bigint FROM config_revision_preview p JOIN prunable pr ON pr.id = p.revision_id CROSS JOIN LATERAL jsonb_each_text(COALESCE(p.fragments, '{}'::jsonb)) ), 0), COALESCE(( SELECT SUM(octet_length(rmp.prefix) + octet_length(COALESCE(rmp.source, '')))::bigint FROM revision_materialized_prefix rmp WHERE rmp.revision_id IN (SELECT id FROM prunable) ), 0), (SELECT COUNT(*)::int FROM freed_snaps), COALESCE((SELECT COUNT(*)::int FROM prefix_snapshot_row psr WHERE psr.snapshot_id IN (SELECT snap_id FROM freed_snaps)), 0), COALESCE(( SELECT SUM(octet_length(psr.prefix) + octet_length(COALESCE(psr.source, '')))::bigint FROM prefix_snapshot_row psr WHERE psr.snapshot_id IN (SELECT snap_id FROM freed_snaps) ), 0)`, tenantID, cutoff.UTC()).Scan( &out.RevisionCount, &revBytes, &previewBytes, &rmpBytes, &out.OrphanSnapshotCount, &out.PrefixRowCount, &snapRowBytes, ) if err != nil { return out, err } out.BytesEstimate = revBytes + previewBytes + rmpBytes + snapRowBytes return out, nil } func (p *Postgres) PruneRevisionsWithStats(tenantID string, cutoff time.Time) (store.RevisionPruneResult, error) { ctx := context.Background() est, err := p.EstimateRevisionPrune(tenantID, cutoff, 0) if err != nil { return store.RevisionPruneResult{}, err } out := store.RevisionPruneResult{BytesEstimate: est.BytesEstimate} where := sqlPrunableRevisionsWhere("$1", "$2") for { cmd, err := p.pool.Exec(ctx, ` DELETE FROM config_revision AS cr WHERE cr.id IN ( SELECT id FROM config_revision WHERE `+where+` ORDER BY created_at ASC LIMIT $3 )`, tenantID, cutoff.UTC(), revisionPruneBatchSize) if err != nil { return out, err } n := int(cmd.RowsAffected()) out.DeletedRevisions += n if n < revisionPruneBatchSize { break } } for { snaps, rows, err := p.pruneUnreferencedPrefixSnapshots(ctx, revisionPruneBatchSize) if err != nil { return out, err } out.DeletedPrefixSnapshots += snaps out.DeletedPrefixRows += rows if snaps < revisionPruneBatchSize { break } } return out, nil } func (p *Postgres) PruneRevisionsBefore(tenantID string, cutoff time.Time) (int, error) { res, err := p.PruneRevisionsWithStats(tenantID, cutoff) if err != nil { return 0, err } return res.DeletedRevisions, nil } func (p *Postgres) pruneUnreferencedPrefixSnapshots(ctx context.Context, batchSize int) (deletedSnapshots int, deletedRows int, err error) { if !prefixSnapshotTableExists(ctx, p.pool) { return 0, 0, nil } var snapCount, rowCount int err = p.pool.QueryRow(ctx, ` WITH doomed AS ( SELECT ps.id FROM prefix_snapshot ps WHERE NOT EXISTS ( SELECT 1 FROM config_revision cr WHERE cr.prefix_snapshot_id = ps.id ) LIMIT $1 ) SELECT (SELECT COUNT(*)::int FROM doomed), (SELECT COUNT(*)::int FROM prefix_snapshot_row psr WHERE psr.snapshot_id IN (SELECT id FROM doomed))`, batchSize).Scan(&snapCount, &rowCount) if err != nil { return 0, 0, err } if snapCount == 0 { return 0, 0, nil } _, err = p.pool.Exec(ctx, ` DELETE FROM prefix_snapshot WHERE id IN ( SELECT ps.id FROM prefix_snapshot ps WHERE NOT EXISTS ( SELECT 1 FROM config_revision cr WHERE cr.prefix_snapshot_id = ps.id ) LIMIT $1 )`, batchSize) if err != nil { return 0, 0, err } return snapCount, rowCount, nil }