EvoBGP/internal/repository/job_audit_integration_test.go

package repository

import (
	"context"
	"errors"
	"evobgp/internal/db"
	"evobgp/internal/store"
	"evobgp/migrations"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"
	"os"
	"sync"
	"testing"
	"time"
)

func TestJobQueueFreshSeedIntegration(t *testing.T) {
	dsn := os.Getenv("EVOBGP_TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("EVOBGP_TEST_DATABASE_URL not set")
	}
	ctx := context.Background()
	admin, err := pgxpool.New(ctx, dsn)
	if err != nil {
		t.Fatal(err)
	}
	defer admin.Close()
	schema := "queue_seed_" + uuid.NewString()
	quoted := `"` + schema + `"`
	if _, err = admin.Exec(ctx, `CREATE SCHEMA `+quoted); err != nil {
		t.Fatal(err)
	}
	defer func() {
		if _, e := admin.Exec(ctx, `DROP SCHEMA `+quoted+` CASCADE`); e != nil {
			t.Error(e)
		}
	}()
	cfg, err := pgxpool.ParseConfig(dsn)
	if err != nil {
		t.Fatal(err)
	}
	cfg.ConnConfig.RuntimeParams["search_path"] = schema
	pool, err := pgxpool.NewWithConfig(ctx, cfg)
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	if err = db.ApplyPostgresMigrations(ctx, pool, migrations.Postgres, "postgres"); err != nil {
		t.Fatal(err)
	}
	p, err := NewPostgres(ctx, pool, true)
	if err != nil {
		t.Fatal(err)
	}
	tenant, cdn, ip, revision, speaker := p.DemoIDs()
	if tenant == "" || tenant == store.SystemJobTenantID || cdn == "" || ip == "" || revision == "" || speaker == "" {
		t.Fatalf("system tenant blocked demo seed: %q %q %q %q %q", tenant, cdn, ip, revision, speaker)
	}
}

func queueIntegration(t *testing.T) (*Postgres, string) {
	t.Helper()
	dsn := os.Getenv("EVOBGP_TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("EVOBGP_TEST_DATABASE_URL not set")
	}
	pool, e := db.OpenPostgresPool(context.Background(), dsn)
	if e != nil {
		t.Fatal(e)
	}
	t.Cleanup(pool.Close)
	tenant := uuid.NewString()
	if _, e = pool.Exec(context.Background(), `INSERT INTO tenant(id,name,slug) VALUES($1,'queue integration',$2)`, tenant, "queue-"+tenant); e != nil {
		t.Fatal(e)
	}
	t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM tenant WHERE id=$1`, tenant) })
	return &Postgres{pool: pool}, tenant
}

func TestJobAuditClaimAndRestartIntegration(t *testing.T) {
	p, tenant := queueIntegration(t)
	ctx := context.Background()
	j, _, e := p.EnqueueJob(ctx, store.JobEnqueue{TenantID: tenant, Kind: "test"})
	if e != nil {
		t.Fatal(e)
	}
	var wg sync.WaitGroup
	results := make(chan []store.JobRecord, 2)
	errs := make(chan error, 2)
	for _, owner := range []string{"replica-a", "replica-b"} {
		wg.Add(1)
		go func(owner string) {
			defer wg.Done()
			items, e := p.ClaimJobs(ctx, owner, 1, time.Minute)
			results <- items
			errs <- e
		}(owner)
	}
	wg.Wait()
	close(results)
	close(errs)
	for e := range errs {
		if e != nil {
			t.Fatal(e)
		}
	}
	var first store.JobRecord
	count := 0
	for items := range results {
		for _, item := range items {
			if item.ID != j.ID {
				t.Fatalf("unexpected job %+v", item)
			}
			first = item
			count++
		}
	}
	if count != 1 {
		t.Fatalf("claims=%d", count)
	}
	if _, e = p.HeartbeatJob(ctx, j.ID, first.OwnerID, first.FencingToken, time.Hour); e != nil {
		t.Fatal(e)
	}
	if items, e := p.ClaimJobs(ctx, "other", 1, time.Minute); e != nil || len(items) != 0 {
		t.Fatalf("stole live lease: %+v %v", items, e)
	}
	if _, e = p.pool.Exec(ctx, `UPDATE job_audit SET lease_until=now()-interval '1 second' WHERE id=$1`, j.ID); e != nil {
		t.Fatal(e)
	}
	pool, e := db.OpenPostgresPool(ctx, os.Getenv("EVOBGP_TEST_DATABASE_URL"))
	if e != nil {
		t.Fatal(e)
	}
	defer pool.Close()
	restarted := &Postgres{pool: pool}
	items, e := restarted.ClaimJobs(ctx, "restarted", 1, time.Minute)
	if e != nil || len(items) != 1 {
		t.Fatalf("restart claim: %+v %v", items, e)
	}
	if items[0].FencingToken <= first.FencingToken {
		t.Fatal("token did not advance")
	}
	first.Status = "succeeded"
	if e = p.FinishJob(ctx, first); !errors.Is(e, store.ErrJobLeaseLost) {
		t.Fatalf("stale finish: %v", e)
	}
	if _, e = p.CancelJob(ctx, tenant, j.ID); e != nil {
		t.Fatal(e)
	}
	items[0].Status = "succeeded"
	if e = restarted.FinishJob(ctx, items[0]); e != nil {
		t.Fatal(e)
	}
	got, e := p.GetJob(ctx, tenant, j.ID)
	if e != nil || got.Status != "cancelled" {
		t.Fatalf("cancel/finish race: %+v %v", got, e)
	}
	listed, _, _, e := restarted.ListJobs(ctx, tenant, "cancelled", "", "", 10)
	if e != nil || len(listed) != 1 {
		t.Fatalf("shared terminal list %+v %v", listed, e)
	}
}

func TestJobAuditIdempotencyRowIntegration(t *testing.T) {
	p, tenant := queueIntegration(t)
	ctx := context.Background()
	key := uuid.NewString()
	in := store.JobEnqueue{TenantID: tenant, Kind: "test", IdempotencyKey: &key, Meta: map[string]any{"input": "same"}}
	var wg sync.WaitGroup
	ids := make(chan string, 8)
	errs := make(chan error, 8)
	for i := 0; i < 8; i++ {
		wg.Add(1)
		go func() { defer wg.Done(); j, _, e := p.EnqueueJob(ctx, in); ids <- j.ID; errs <- e }()
	}
	wg.Wait()
	close(ids)
	close(errs)
	for e := range errs {
		if e != nil {
			t.Fatal(e)
		}
	}
	id := ""
	for got := range ids {
		if id == "" {
			id = got
		}
		if got != id {
			t.Fatalf("different IDs %s / %s", id, got)
		}
	}
	items, e := p.ClaimJobs(ctx, "replica", 1, time.Minute)
	if e != nil || len(items) != 1 {
		t.Fatalf("claim %+v %v", items, e)
	}
	items[0].Status = "succeeded"
	items[0].Meta = map[string]any{"result": "done"}
	if e = p.FinishJob(ctx, items[0]); e != nil {
		t.Fatal(e)
	}
	got, created, e := p.EnqueueJob(ctx, in)
	if e != nil || created || got.ID != id || got.Status != "succeeded" {
		t.Fatalf("terminal replay %+v %v %v", got, created, e)
	}
	in.Meta = map[string]any{"input": "changed"}
	if _, _, e = p.EnqueueJob(ctx, in); !errors.Is(e, store.ErrJobConflict) {
		t.Fatalf("changed request accepted: %v", e)
	}
}

func TestJobQueueRetryBoundsAndFailureIntegration(t *testing.T) {
	p, tenant := queueIntegration(t)
	ctx := context.Background()
	t.Setenv("EVOBGP_JOB_MAX_TENANT_PENDING", "1")
	j, _, e := p.EnqueueJob(ctx, store.JobEnqueue{TenantID: tenant, Kind: "test"})
	if e != nil {
		t.Fatal(e)
	}
	if _, _, e = p.EnqueueJob(ctx, store.JobEnqueue{TenantID: tenant, Kind: "test"}); !errors.Is(e, store.ErrJobQueueFull) {
		t.Fatalf("unbounded queue: %v", e)
	}
	if _, e = p.pool.Exec(ctx, `UPDATE job_audit SET attempt=3 WHERE id=$1`, j.ID); e != nil {
		t.Fatal(e)
	}
	if e = p.ReapExpiredJobs(ctx); e != nil {
		t.Fatal(e)
	}
	items, e := p.ClaimJobs(ctx, "worker", 1, time.Minute)
	if e != nil || len(items) != 0 {
		t.Fatalf("exhausted claim %+v %v", items, e)
	}
	got, e := p.GetJob(ctx, tenant, j.ID)
	if e != nil || got.Status != "failed" {
		t.Fatalf("exhausted status %+v %v", got, e)
	}
	if _, e = p.RetryJob(ctx, tenant, j.ID); e != nil {
		t.Fatal(e)
	}
	items, e = p.ClaimJobs(ctx, "worker", 1, time.Minute)
	if e != nil || len(items) != 1 || items[0].Attempt != 1 {
		t.Fatalf("operator retry %+v %v", items, e)
	}
	if e = p.ReleaseJob(ctx, j.ID, items[0].OwnerID, items[0].FencingToken); e != nil {
		t.Fatal(e)
	}
	items, e = p.ClaimJobs(ctx, "worker-2", 1, time.Minute)
	if e != nil || len(items) != 1 || items[0].Attempt != 1 {
		t.Fatalf("shutdown consumed retry %+v %v", items, e)
	}
	failedPool, e := db.OpenPostgresPool(ctx, os.Getenv("EVOBGP_TEST_DATABASE_URL"))
	if e != nil {
		t.Fatal(e)
	}
	failedPool.Close()
	broken := &Postgres{pool: failedPool}
	if accepted, created, e := broken.EnqueueJob(ctx, store.JobEnqueue{TenantID: tenant, Kind: "test"}); e == nil || created || accepted.ID != "" {
		t.Fatalf("accepted on failed DB: %+v %v %v", accepted, created, e)
	}
}