EvoBGP/internal/repository/module_snapshot.go

package repository

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"sync"
	"time"

	"evobgp/internal/store"

	"github.com/jackc/pgx/v5"
)

// moduleSnapshotRowExistsOnce caches the schema check for the process lifetime.
// The table is created by migrations and never disappears at runtime; DB errors are
// not cached so a transient outage falls back to the JSON path only once.
var moduleSnapshotRowExistsOnce sync.Once

var moduleSnapshotRowExistsCached bool

func moduleSnapshotRowTableExists(ctx context.Context, q queryRower) bool {
	moduleSnapshotRowExistsOnce.Do(func() {
		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)
		// A query error means the backend is unreachable or information_schema is hidden;
		// treat as "absent" so callers fall back to prefixes_json, but retry next call.
		if err != nil {
			moduleSnapshotRowExistsOnce = sync.Once{}
			return
		}
		moduleSnapshotRowExistsCached = true
	})
	return moduleSnapshotRowExistsCached
}

func (p *Postgres) GetModulePrefixSnapshot(tenantID, moduleID string) (*store.ModulePrefixSnapshot, bool, error) {
	ctx := context.Background()
	var inputHash string
	var collectedAt time.Time
	err := p.pool.QueryRow(ctx, `
		SELECT input_hash, collected_at
		FROM module_prefix_snapshot
		WHERE tenant_id = $1 AND module_id = $2`,
		tenantID, moduleID).Scan(&inputHash, &collectedAt)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, false, nil
		}
		return nil, false, err
	}
	var prefixes []store.PrefixRow
	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{
		InputHash:   inputHash,
		CollectedAt: collectedAt.UTC(),
		Prefixes:    prefixes,
	}, true, nil
}

func (p *Postgres) SetModulePrefixSnapshot(tenantID, moduleID, inputHash string, prefixes []store.PrefixRow) error {
	if strings.TrimSpace(tenantID) == "" || strings.TrimSpace(moduleID) == "" || strings.TrimSpace(inputHash) == "" {
		return store.ErrInvalidInput
	}
	ctx := context.Background()
	tx, err := p.pool.Begin(ctx)
	if err != nil {
		return err
	}
	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`,
		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
		}
		if len(prefixes) > 0 {
			_, err = tx.CopyFrom(
				ctx,
				pgx.Identifier{"module_prefix_snapshot_row"},
				[]string{"tenant_id", "module_id", "ord", "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{tenantID, moduleID, i, strings.TrimSpace(pr.Prefix), comm, src}, nil
				}),
			)
			if 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 {
	ctx := context.Background()
	_, err := p.pool.Exec(ctx, `
		DELETE FROM module_prefix_snapshot WHERE tenant_id = $1 AND module_id = $2`,
		tenantID, moduleID)
	return err
}

func (p *Postgres) SetModuleInputHash(tenantID, moduleID, hash string) error {
	ctx := context.Background()
	tag, err := p.pool.Exec(ctx, `
		UPDATE module SET input_hash = $3
		WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL`,
		moduleID, tenantID, hash)
	if err != nil {
		return err
	}
	if tag.RowsAffected() == 0 {
		return store.ErrNotFound
	}
	return nil
}

func (p *Postgres) LockModuleSnapshot(tenantID, moduleID string) func() {
	ctx := context.Background()
	conn, err := p.pool.Acquire(ctx)
	if err != nil {
		return func() {}
	}
	key := strings.TrimSpace(tenantID) + "\x00" + strings.TrimSpace(moduleID)
	if _, err := conn.Exec(ctx, `SELECT pg_advisory_lock(hashtext($1))`, key); err != nil {
		conn.Release()
		return func() {}
	}
	return func() {
		_, _ = conn.Exec(ctx, `SELECT pg_advisory_unlock(hashtext($1))`, key)
		conn.Release()
	}
}