ITWorx-Pulse-Public/internal/agentstore/postgres_integration_test.go

package agentstore

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"testing"
	"time"

	"github.com/itworx/pulse/internal/database"
)

// TestPostgreSQLSnapshotRoundTrip exercises the real table created by migration 0016. It
// skips unless PULSE_TEST_DATABASE_URL points at a disposable database.
func TestPostgreSQLSnapshotRoundTrip(t *testing.T) {
	dsn := os.Getenv("PULSE_TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("PULSE_TEST_DATABASE_URL is not set")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	pool, err := database.NewPool(ctx, database.Config{URL: dsn})
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	if err := database.Migrate(ctx, pool); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `DELETE FROM agent_snapshots WHERE agent_id = 'integration-agent'`); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `DELETE FROM capacity_samples WHERE source_id = 'integration-agent'`); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() {
		cleanup, cancelCleanup := context.WithTimeout(context.Background(), 15*time.Second)
		defer cancelCleanup()
		_, _ = pool.Exec(cleanup, `DELETE FROM agent_snapshots WHERE agent_id = 'integration-agent'`)
		_, _ = pool.Exec(cleanup, `DELETE FROM capacity_samples WHERE source_id = 'integration-agent'`)
	})

	now := time.Now().UTC().Truncate(time.Microsecond)
	store := PostgresStore{Pool: pool, Clock: func() time.Time { return now }}
	readOwn := func() Snapshot {
		t.Helper()
		stored := Snapshot{AgentID: "integration-agent", Capability: CapabilityPools}
		var payload []byte
		if err := pool.QueryRow(ctx, `SELECT observed_at,received_at,payload FROM agent_snapshots WHERE agent_id=$1 AND capability=$2`, stored.AgentID, stored.Capability).Scan(&stored.ObservedAt, &stored.ReceivedAt, &payload); err != nil {
			t.Fatal(err)
		}
		stored.Payload = payload
		return stored
	}
	if _, err := store.Latest(ctx, CapabilityPools); err != nil && !errors.Is(err, ErrNoSnapshot) {
		t.Fatalf("unexpected error reading an empty capability: %v", err)
	}

	first := Snapshot{AgentID: "integration-agent", Capability: CapabilityPools, ObservedAt: now.Add(-30 * time.Second), ReceivedAt: now.Add(72 * time.Hour), Payload: json.RawMessage(`{"pools":[{"id":"cache","name":"Cache","usedBytes":100,"usableBytes":1000}],"generation":1}`)}
	if err := store.Put(ctx, first); err != nil {
		t.Fatal(err)
	}
	stored := readOwn()
	if !stored.ReceivedAt.Equal(now) {
		t.Fatalf("received at = %s, want the store clock %s", stored.ReceivedAt, now)
	}
	if !stored.ObservedAt.Equal(first.ObservedAt) {
		t.Fatalf("observed at = %s, want %s", stored.ObservedAt, first.ObservedAt)
	}

	newer := first
	newer.ObservedAt = now.Add(-5 * time.Second)
	newer.Payload = json.RawMessage(`{"pools":[{"id":"cache","name":"Cache","usedBytes":200,"usableBytes":1000}],"generation":2}`)
	if err := store.Put(ctx, newer); err != nil {
		t.Fatal(err)
	}
	older := first
	older.ObservedAt = now.Add(-120 * time.Second)
	older.Payload = json.RawMessage(`{"pools":[{"id":"cache","name":"Cache","usedBytes":50,"usableBytes":1000}],"generation":3}`)
	if err := store.Put(ctx, older); err != nil {
		t.Fatal(err)
	}
	stored = readOwn()
	var decoded struct {
		Generation int `json:"generation"`
	}
	if err := json.Unmarshal(stored.Payload, &decoded); err != nil {
		t.Fatal(err)
	}
	if decoded.Generation != 2 {
		t.Fatalf("a delayed retry must not resurrect superseded telemetry, got generation %d", decoded.Generation)
	}

	var rows int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM agent_snapshots WHERE agent_id = 'integration-agent'`).Scan(&rows); err != nil {
		t.Fatal(err)
	}
	if rows != 1 {
		t.Fatalf("agent snapshot rows = %d, want exactly one per (agent, capability)", rows)
	}
	var sampleRows int
	var usedBytes int64
	if err := pool.QueryRow(ctx, `SELECT count(*),max(used_bytes) FROM capacity_samples WHERE source_id='integration-agent' AND entity_kind='pool' AND entity_id='cache'`).Scan(&sampleRows, &usedBytes); err != nil {
		t.Fatal(err)
	}
	if sampleRows != 1 || usedBytes != 200 {
		t.Fatalf("six-hour bucket was not idempotent or accepted an older retry: rows=%d used=%d", sampleRows, usedBytes)
	}
}