ITWorx-Pulse-Public/internal/alert/state_repository_integration_test.go

package alert

import (
	"context"
	"os"
	"sync"
	"testing"
	"time"

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

func TestPostgreSQLAlertStateLifecycleAndIdempotence(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(), 45*time.Second)
	defer cancel()
	pool, err := database.NewPool(ctx, database.Config{URL: dsn, MaxConns: 8, MinConns: 1})
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	if err := database.Migrate(ctx, pool); err != nil {
		t.Fatal(err)
	}
	if err := database.Migrate(ctx, pool); err != nil {
		t.Fatal(err)
	}
	document, registry := validDocument(t)
	document.Enabled = true
	rules := Repository{Pool: pool, Registry: registry}
	created, version, err := rules.Create(ctx, "state-integration", document, "state test")
	if err != nil {
		t.Fatal(err)
	}
	store := StateRepository{Pool: pool}
	policy := Policy{PendingSeconds: document.PendingSeconds, ResolveSeconds: document.ResolveSeconds, UnknownBehavior: document.UnknownBehavior}
	base := time.Date(2026, time.January, 2, 12, 0, 0, 0, time.UTC)
	first, firstOccurrence, duplicate, err := store.ApplyObservation(ctx, StateInput{RuleID: created.ID, RuleVersionID: version.ID, Fingerprint: "host:test", Policy: policy, Observation: observation(base, "slot-1", true)})
	if err != nil {
		t.Fatal(err)
	}
	if duplicate || first.State != StatePending || firstOccurrence.To != StatePending {
		t.Fatalf("unexpected first state: %#v %#v", first, firstOccurrence)
	}
	replayed, replayOccurrence, duplicate, err := store.ApplyObservation(ctx, StateInput{RuleID: created.ID, RuleVersionID: version.ID, Fingerprint: "host:test", Policy: policy, Observation: observation(base, "slot-1", true)})
	if err != nil {
		t.Fatal(err)
	}
	if !duplicate || replayed.Revision != first.Revision || replayOccurrence.ID != firstOccurrence.ID {
		t.Fatalf("replay was not idempotent: %#v %#v", replayed, replayOccurrence)
	}
	firing, _, _, err := store.ApplyObservation(ctx, StateInput{RuleID: created.ID, RuleVersionID: version.ID, Fingerprint: "host:test", Policy: policy, Observation: observation(base.Add(60*time.Second), "slot-2", true)})
	if err != nil {
		t.Fatal(err)
	}
	if firing.State != StateFiring {
		t.Fatalf("pending did not fire: %#v", firing)
	}
	acknowledged, _, _, err := store.Acknowledge(ctx, firing.ID, "operator", "ack-1", base.Add(61*time.Second))
	if err != nil {
		t.Fatal(err)
	}
	if acknowledged.State != StateAcknowledged {
		t.Fatalf("acknowledgement failed: %#v", acknowledged)
	}
	stillFiring, _, _, err := store.ApplyObservation(ctx, StateInput{RuleID: created.ID, RuleVersionID: version.ID, Fingerprint: "host:test", Policy: policy, Observation: observation(base.Add(62*time.Second), "slot-3", true)})
	if err != nil {
		t.Fatal(err)
	}
	if stillFiring.State != StateAcknowledged {
		t.Fatalf("acknowledged firing alert changed state: %#v", stillFiring)
	}
	unknownObservation := observation(base.Add(90*time.Second), "slot-4", false)
	unknownObservation.Unknown = true
	unknown, _, _, err := store.ApplyObservation(ctx, StateInput{RuleID: created.ID, RuleVersionID: version.ID, Fingerprint: "host:test", Policy: policy, Observation: unknownObservation})
	if err != nil {
		t.Fatal(err)
	}
	if unknown.State != StateUnknown || unknown.RetainedState != StateAcknowledged {
		t.Fatalf("unknown state lost acknowledgement context: %#v", unknown)
	}
	restarted := StateRepository{Pool: pool}
	loaded, err := restarted.GetInstance(ctx, unknown.ID)
	if err != nil {
		t.Fatal(err)
	}
	if loaded.State != StateUnknown || loaded.LastKnownAt == nil || loaded.LastValue == nil {
		t.Fatalf("restart did not preserve state: %#v", loaded)
	}
	occurrences, err := restarted.ListOccurrences(ctx, unknown.ID, 20)
	if err != nil {
		t.Fatal(err)
	}
	if len(occurrences) != 5 {
		t.Fatalf("occurrence count = %d, want 5", len(occurrences))
	}

	missingVersion := StateInput{RuleID: created.ID, RuleVersionID: NewID(), Fingerprint: "rollback", Policy: policy, Observation: observation(base, "rollback", true)}
	if _, _, _, err := store.ApplyObservation(ctx, missingVersion); err == nil {
		t.Fatal("missing foreign key did not fail")
	}
	if _, err := pool.Exec(ctx, `SELECT 1 FROM alert_instances WHERE rule_id=$1 AND fingerprint=$2`, created.ID, "rollback"); err != nil {
		t.Fatal(err)
	}
}

func TestPostgreSQLAlertStateCoordinatesOverlappingWrites(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(), 45*time.Second)
	defer cancel()
	pool, err := database.NewPool(ctx, database.Config{URL: dsn, MaxConns: 8, MinConns: 1})
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	if err := database.Migrate(ctx, pool); err != nil {
		t.Fatal(err)
	}
	document, registry := validDocument(t)
	rules := Repository{Pool: pool, Registry: registry}
	created, version, err := rules.Create(ctx, "state-concurrency", document, "state concurrency")
	if err != nil {
		t.Fatal(err)
	}
	store := StateRepository{Pool: pool}
	input := StateInput{RuleID: created.ID, RuleVersionID: version.ID, Fingerprint: "host:concurrent", Policy: Policy{PendingSeconds: 0, ResolveSeconds: 0, UnknownBehavior: UnknownRetain}, Observation: observation(time.Date(2026, time.January, 3, 12, 0, 0, 0, time.UTC), "same-slot", true)}
	const workers = 8
	results := make(chan bool, workers)
	errorsCh := make(chan error, workers)
	var group sync.WaitGroup
	for i := 0; i < workers; i++ {
		group.Add(1)
		go func() {
			defer group.Done()
			_, _, duplicate, err := store.ApplyObservation(ctx, input)
			if err != nil {
				errorsCh <- err
				return
			}
			results <- duplicate
		}()
	}
	group.Wait()
	close(results)
	close(errorsCh)
	for err := range errorsCh {
		t.Fatal(err)
	}
	createdCount := 0
	for duplicate := range results {
		if !duplicate {
			createdCount++
		}
	}
	if createdCount != 1 {
		t.Fatalf("non-idempotent concurrent writes = %d, want 1", createdCount)
	}
	var occurrenceCount int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM alert_occurrences WHERE instance_id=(SELECT id FROM alert_instances WHERE rule_id=$1 AND fingerprint=$2)`, created.ID, input.Fingerprint).Scan(&occurrenceCount); err != nil {
		t.Fatal(err)
	}
	if occurrenceCount != 1 {
		t.Fatalf("occurrences = %d, want 1", occurrenceCount)
	}
}