access-engine/internal/emission/delivery_test.go

package emission_test

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/netkingdom/access-engine/internal/emission"
	"github.com/netkingdom/access-engine/pkg/api"
)

func TestHeartbeatScheduleSurvivesRestart(t *testing.T) {
	dir := t.TempDir()
	now := time.Date(2026, 10, 4, 0, 0, 0, 0, time.UTC)
	o, err := emission.Open(dir, "flex-auth.test")
	if err != nil {
		t.Fatal(err)
	}
	o.SetClock(func() time.Time { return now })
	if err := o.AppendDueHeartbeats(); err != nil {
		t.Fatal(err)
	}
	if err := o.AppendDueHeartbeats(); err != nil {
		t.Fatal(err)
	}
	o.Close()
	o, err = emission.Open(dir, "flex-auth.test")
	if err != nil {
		t.Fatal(err)
	}
	defer o.Close()
	o.SetClock(func() time.Time { return now })
	if err := o.AppendDueHeartbeats(); err != nil {
		t.Fatal(err)
	}
	if got := len(readEvents(t, dir)); got != 4 {
		t.Fatalf("heartbeats after restart = %d", got)
	}
	seen := map[string]bool{}
	for _, e := range readEvents(t, dir) {
		if e["type"] != emission.HeartbeatType || e["id"] != e["correlation_id"] {
			t.Fatalf("invalid heartbeat: %v", e)
		}
		if _, err := time.Parse(time.RFC3339Nano, e["occurred_at"].(string)); err != nil {
			t.Fatal(err)
		}
		data := e["data"].(map[string]any)
		seen[data["class"].(string)] = true
	}
	for _, class := range emission.RareClasses() {
		if !seen[class] {
			t.Fatalf("no heartbeat for %s", class)
		}
	}
	now = now.Add(24 * time.Hour)
	o.NoteReleasedUncommitted(api.DecisionEffectDeny)
	if err := o.AppendDueHeartbeats(); err != nil {
		t.Fatal(err)
	}
	if got := len(readEvents(t, dir)); got != 8 {
		t.Fatalf("daily heartbeats = %d", got)
	}
	for _, event := range readEvents(t, dir)[4:] {
		data := event["data"].(map[string]any)
		if data["class"] == "flex-auth.decision.deny" && data["assertion"] != "released-uncommitted-finding" {
			t.Fatal("known emission failure concealed by nothing-to-report heartbeat")
		}
	}
}

func TestDrainRetainsRejectionsAndResumesWithStableIDs(t *testing.T) {
	for _, rejection := range []int{400, 401, 403, 500, 302, 204} {
		t.Run(http.StatusText(rejection), func(t *testing.T) {
			dir := t.TempDir()
			tokenFile := filepath.Join(dir, "sender-token")
			if err := os.WriteFile(tokenFile, []byte("test-token"), 0600); err != nil {
				t.Fatal(err)
			}
			o, err := emission.Open(dir, "flex-auth.test")
			if err != nil {
				t.Fatal(err)
			}
			if err := o.AppendBatch([]api.DecisionEnvelope{deny("decision:1", "tenant:a"), deny("decision:2", "tenant:a")}); err != nil {
				t.Fatal(err)
			}
			var ids []string
			status := rejection
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				var event emission.Event
				if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
					t.Error(err)
				}
				if r.URL.Path != "/v1/events" || r.Header.Get("Idempotency-Key") != event.ID || r.Header.Get("Authorization") != "Bearer test-token" {
					t.Errorf("bad delivery headers/path")
				}
				ids = append(ids, event.ID)
				w.Header().Set("Location", "/redirect-must-not-be-followed")
				w.WriteHeader(status)
				w.Write([]byte("sensitive receiver payload"))
			}))
			defer server.Close()
			cfg := emission.DeliveryConfig{URL: server.URL + "/v1/events", TokenFile: tokenFile}
			if n, err := o.Drain(context.Background(), cfg, 100); n != 0 || err == nil || strings.Contains(err.Error(), "sensitive") {
				t.Fatalf("rejection advanced or exposed response: %d %v", n, err)
			}
			if o.Status().Pending != 2 {
				t.Fatal("rejected event lost")
			}
			status = 202
			if n, err := o.Drain(context.Background(), cfg, 1); n != 1 || err != nil {
				t.Fatalf("accepted: %d %v", n, err)
			}
			o.Close()
			o, err = emission.Open(dir, "flex-auth.test")
			if err != nil {
				t.Fatal(err)
			}
			defer o.Close()
			if o.Status().Pending != 1 {
				t.Fatal("cursor did not survive restart")
			}
			status = 200
			if n, err := o.Drain(context.Background(), cfg, 100); n != 1 || err != nil {
				t.Fatalf("duplicate: %d %v", n, err)
			}
			if len(ids) != 3 || ids[0] != ids[1] || ids[1] == ids[2] {
				t.Fatalf("unstable/reused event IDs: %v", ids)
			}
			if o.Status().Pending != 0 {
				t.Fatal("pending after drain")
			}
		})
	}
}

func TestDrainAmbiguousAcceptanceReplaysAndTokenRotates(t *testing.T) {
	dir := t.TempDir()
	tokenFile := filepath.Join(dir, "token")
	os.WriteFile(tokenFile, []byte("first"), 0600)
	o, err := emission.Open(dir, "flex-auth.test")
	if err != nil {
		t.Fatal(err)
	}
	defer o.Close()
	o.Append(deny("decision:1", "tenant:a"))
	var ids, tokens []string
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		ids = append(ids, r.Header.Get("Idempotency-Key"))
		tokens = append(tokens, r.Header.Get("Authorization"))
		w.WriteHeader(202)
	}))
	defer server.Close()
	cfg := emission.DeliveryConfig{URL: server.URL + "/v1/events", TokenFile: tokenFile}
	// Receiver accepts, but persisting the cursor fails. Retain and retry.
	os.Mkdir(filepath.Join(dir, "drain.json"), 0700)
	if n, err := o.Drain(context.Background(), cfg, 1); n != 0 || err == nil {
		t.Fatalf("cursor failure: %d %v", n, err)
	}
	os.Remove(filepath.Join(dir, "drain.json"))
	os.WriteFile(tokenFile, []byte("rotated"), 0600)
	if n, err := o.Drain(context.Background(), cfg, 1); n != 1 || err != nil {
		t.Fatalf("retry: %d %v", n, err)
	}
	if len(ids) != 2 || ids[0] != ids[1] || tokens[1] != "Bearer rotated" {
		t.Fatalf("retry/rotation: %v %v", ids, tokens)
	}
}

func TestOutboxReportsLagAndRefusesSourceReuse(t *testing.T) {
	dir := t.TempDir()
	o, err := emission.Open(dir, "flex-auth.test")
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC()
	o.SetClock(func() time.Time { return now })
	o.Append(deny("decision:1", "tenant:a"))
	if o.Status().LagExceeded {
		t.Fatal("fresh event treated as lag finding")
	}
	now = now.Add(2 * time.Hour)
	if !o.Status().LagExceeded {
		t.Fatal("overdue event not reported")
	}
	o.Close()
	if other, err := emission.Open(dir, "flex-auth.other"); err == nil {
		other.Close()
		t.Fatal("source changed on existing outbox")
	}
}

func TestSlowReceiverDoesNotBlockDecisionCommit(t *testing.T) {
	dir := t.TempDir()
	tokenFile := filepath.Join(dir, "token")
	if err := os.WriteFile(tokenFile, []byte("local-test-token"), 0600); err != nil {
		t.Fatal(err)
	}
	o, err := emission.Open(dir, "flex-auth.test")
	if err != nil {
		t.Fatal(err)
	}
	defer o.Close()
	if err := o.Append(deny("decision:1", "tenant:a")); err != nil {
		t.Fatal(err)
	}
	arrived, release := make(chan struct{}), make(chan struct{})
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		close(arrived)
		<-release
		w.WriteHeader(202)
	}))
	defer server.Close()
	drained := make(chan error, 1)
	go func() {
		_, err := o.Drain(context.Background(), emission.DeliveryConfig{URL: server.URL + "/v1/events", TokenFile: tokenFile}, 1)
		drained <- err
	}()
	defer func() {
		close(release)
		if err := <-drained; err != nil {
			t.Error(err)
		}
	}()
	<-arrived
	committed := make(chan error, 1)
	go func() { committed <- o.Append(deny("decision:2", "tenant:a")) }()
	select {
	case err := <-committed:
		if err != nil {
			t.Fatal(err)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("remote HTTP holds the decision commit lock")
	}
	// Deferred release lets the drain finish even on test failure. Assert the
	// pending count while HTTP is still blocked rather than waiting for it.
	if o.Status().Pending != 2 {
		t.Fatal("new commit lost during delivery")
	}
}