creator-hub/internal/creator/notice_polling_test.go

package creator

import (
	"errors"
	"fmt"
	"testing"
	"time"
)

func TestNotificationCheckpointValidation(t *testing.T) {
	now := time.Now().UTC()
	valid := ListenerDelivery{DeliveryID: "scan", Kind: "checkpoint", BoundaryAt: &now, CheckedAt: &now, Checkpoints: map[string]string{"960": "9007199254740993"}}
	if err := valid.Validate(); err != nil {
		t.Fatal(err)
	}
	for _, d := range []ListenerDelivery{
		{DeliveryID: "scan", Kind: "checkpoint"},
		{DeliveryID: "scan", Kind: "checkpoint", BoundaryAt: &now, CheckedAt: &now, Checkpoints: map[string]string{"invalid": "1"}},
		{DeliveryID: "scan", Kind: "checkpoint", BoundaryAt: &now, CheckedAt: &now, Checkpoints: map[string]string{"960": "bad"}},
	} {
		if !errors.Is(d.Validate(), ErrInvalid) {
			t.Fatalf("accepted %+v", d)
		}
	}
}

func TestNotificationCheckpointTransactionAndRestart(t *testing.T) {
	store, accounts, ctx := openCreatorIntegrationStore(t)
	id := createIntegrationAccount(t, ctx, accounts, fmt.Sprint(time.Now().UnixNano()))
	profile, err := store.GetAccountProfile(ctx, id)
	if err != nil {
		t.Fatal(err)
	}
	if _, err = store.RecordVerifiedLoginResult(ctx, id, profile.PlatformAccountKey); err != nil {
		t.Fatal(err)
	}
	state, err := store.SetListenerEnabled(ctx, id, true)
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC().Truncate(time.Microsecond)
	checkpoint := ListenerDelivery{DeliveryID: "scan", Kind: "checkpoint", BoundaryAt: &now, CheckedAt: &now, Checkpoints: map[string]string{"960": "9007199254740993"}}
	items := []ListenerDelivery{{DeliveryID: "event", Kind: "notice", Notice: &ListenerNotice{EventKey: "9007199254740993", EventType: "follow", PlatformEventAt: &now}}, checkpoint}
	if err = store.RecordListenerDeliveries(ctx, id, state.Generation, items); err != nil {
		t.Fatal(err)
	}
	profile, err = store.GetAccountProfile(ctx, id)
	if err != nil || profile.ListenerLastPollAt == nil || !profile.ListenerLastPollAt.Equal(now) {
		t.Fatalf("account does not expose last successful scan: %+v %v", profile, err)
	}
	states, err := store.ListListenerStates(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if states[0].Checkpoints["960"] != "9007199254740993" || !states[0].BoundaryAt.Equal(now) || !states[0].LastPollAt.Equal(now) || states[0].Status != "ready" {
		t.Fatalf("checkpoint not durable: %+v", states[0])
	}
	later := now.Add(time.Minute)
	bad := checkpoint
	bad.BoundaryAt = &later
	bad.Checkpoints = map[string]string{"960": "2"}
	if err = store.RecordListenerDeliveries(ctx, id, state.Generation, []ListenerDelivery{bad}); !errors.Is(err, ErrConflict) {
		t.Fatalf("boundary replaced: %v", err)
	}
	if err = store.RecordListenerDeliveries(ctx, id, state.Generation, items); err != nil {
		t.Fatal(err)
	}
	page, err := store.ListAggregatedEventsPage(ctx, EventAggregationFilter{}, 1, 25)
	if err != nil || page.Total != 1 {
		t.Fatalf("duplicate inserted: %+v %v", page, err)
	}
	gap := checkpoint
	gap.Reason = "checkpoint disappeared"
	if err = store.RecordListenerDeliveries(ctx, id, state.Generation, []ListenerDelivery{gap}); err != nil {
		t.Fatal(err)
	}
	states, err = store.ListListenerStates(ctx)
	if err != nil || states[0].Status != "gap" {
		t.Fatalf("missing boundary hidden: %+v %v", states, err)
	}
	off, err := store.SetListenerEnabled(ctx, id, false)
	if err != nil {
		t.Fatal(err)
	}
	if err = store.RecordListenerDeliveries(ctx, id, state.Generation, items); !errors.Is(err, ErrConflict) {
		t.Fatal(err)
	}
	on, err := store.SetListenerEnabled(ctx, id, true)
	if err != nil {
		t.Fatal(err)
	}
	if on.Generation == off.Generation || on.BoundaryAt != nil || len(on.Checkpoints) != 0 || on.LastPollAt != nil {
		t.Fatalf("new enable inherited old boundary: %+v", on)
	}
}