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