diff --git a/.codescene/code-health-rules.json b/.codescene/code-health-rules.json new file mode 100644 index 00000000..6279b46f --- /dev/null +++ b/.codescene/code-health-rules.json @@ -0,0 +1,10 @@ +{ + "rule_sets": [ + { + "matching_content_path": "**/*_test.go", + "rules": [ + { "name": "String Heavy Function Arguments", "weight": 0.0 } + ] + } + ] +} diff --git a/.env.example b/.env.example index a9e158ae..9500851c 100644 --- a/.env.example +++ b/.env.example @@ -31,6 +31,9 @@ ALPHONE_TEST_DATABASE_MAX_AGE=1h # How long a server process keeps a tenant's fields before reading them again, 1m when unset. # ALPHONE_TENANTS_REFRESH=1m +# The most contact addresses a plugin matches in one call, and reads in one page of new ones, 100 when unset. +# ALPHONE_CONTACT_MATCH_MAX=100 + # Rows a graph list answers when the caller names none, core and plugin lists alike, 50 when unset. # ALPHONE_GRAPH_PAGE_SIZE=50 # The most rows one graph list answers, core and plugin lists alike, 200 when unset. diff --git a/cmd/alphone/compose.go b/cmd/alphone/compose.go index 1c975785..8ae51e1b 100644 --- a/cmd/alphone/compose.go +++ b/cmd/alphone/compose.go @@ -78,17 +78,20 @@ func compose(ctx context.Context, cfg composeConfig, plugins func(sdk.Deps) ([]s } resolver := contact.NewResolver(built.contacts, contact.WithEvents(built.events)) built.registered, built.failed = plugins(sdk.Deps{ - DatabaseURL: cfg.databaseURL, - PublicURL: cfg.mail.publicURL, - MachineGrace: cfg.machineGrace, - TenantsHeld: cfg.tenants.held, - TenantsRefresh: cfg.tenants.refresh, - Pages: sdk.PageBounds{Size: cfg.paging.Size, Cap: cfg.paging.Cap}, - Resolver: resolverBridge{resolver: resolver}, - Contacts: directoryBridge{resolver: resolver}, - Events: pluginPublisher{publisher: built.events, logger: cfg.logger}, - Getenv: cfg.getenv, - Env: settingsEnv(cfg.getenv), + DatabaseURL: cfg.databaseURL, + PublicURL: cfg.mail.publicURL, + MachineGrace: cfg.machineGrace, + TenantsHeld: cfg.tenants.held, + TenantsRefresh: cfg.tenants.refresh, + Pages: sdk.PageBounds{Size: cfg.paging.Size, Cap: cfg.paging.Cap}, + Resolver: resolverBridge{resolver: resolver}, + Contacts: directoryBridge{resolver: resolver}, + Matcher: matcherBridge{matcher: contact.NewMatcher(built.contacts, cfg.contactMatchMax)}, + ContactMatchMax: cfg.contactMatchMax, + Identities: identityFeedBridge{feed: contact.NewFeed(built.contacts, cfg.contactMatchMax)}, + Events: pluginPublisher{publisher: built.events, logger: cfg.logger}, + Getenv: cfg.getenv, + Env: settingsEnv(cfg.getenv), }) if err := declareRoles(cfg.roles, built.registered); err != nil { return built, fmt.Errorf("declare plugin roles: %w", err) diff --git a/cmd/alphone/compose_test.go b/cmd/alphone/compose_test.go index 3978116d..27125a93 100644 --- a/cmd/alphone/compose_test.go +++ b/cmd/alphone/compose_test.go @@ -213,6 +213,22 @@ func TestComposeHandsThePluginsTheConfiguredPageBounds(t *testing.T) { } } +func TestComposeHandsThePluginsTheConfiguredContactMatchBound(t *testing.T) { + t.Parallel() + + var handed sdk.Deps + + built, err := compose(t.Context(), composeConfigOf(t, filledSettings()), capturingPlugins(&handed)) + + if err != nil { + t.Fatalf("compose() error = %v, want nil", err) + } + abandonAtEnd(t, built) + if handed.ContactMatchMax != 40 { + t.Errorf("Deps.ContactMatchMax = %d, want the configured 40", handed.ContactMatchMax) + } +} + func TestComposeWiresTheFieldProviders(t *testing.T) { t.Parallel() diff --git a/cmd/alphone/contactmatch_exec_test.go b/cmd/alphone/contactmatch_exec_test.go new file mode 100644 index 00000000..521a46c7 --- /dev/null +++ b/cmd/alphone/contactmatch_exec_test.go @@ -0,0 +1,34 @@ +// SPDX-License-Identifier: Elastic-2.0 + +package main + +import ( + "bytes" + "errors" + "os/exec" + "strings" + "testing" + + "github.com/gopherium/framework/gonsole/testkit" +) + +func TestMainBinaryRefusesAContactMatchBoundOfZero(t *testing.T) { + t.Parallel() + + binary, env := testkit.CoverBinary(t, "ALPHONE_", "alphone") + var stderr bytes.Buffer + cmd := exec.Command(binary, "serve") + cmd.Dir = t.TempDir() + cmd.Env = append(env, "ALPHONE_DATABASE_URL="+unreachableDatabaseURL, "ALPHONE_CONTACT_MATCH_MAX=0") + cmd.Stderr = &stderr + + err := cmd.Run() + + var exitErr *exec.ExitError + if !errors.As(err, &exitErr) || exitErr.ExitCode() != 1 { + t.Fatalf("alphone serve with a match bound of zero: %v, want exit code 1", err) + } + if want := `ALPHONE_CONTACT_MATCH_MAX: must stand above zero, got "0"`; !strings.Contains(stderr.String(), want) { + t.Errorf("stderr = %q, want it to name %s", stderr.String(), want) + } +} diff --git a/cmd/alphone/identitywiring_test.go b/cmd/alphone/identitywiring_test.go new file mode 100644 index 00000000..db32a7de --- /dev/null +++ b/cmd/alphone/identitywiring_test.go @@ -0,0 +1,216 @@ +// SPDX-License-Identifier: Elastic-2.0 + +package main + +import ( + "context" + "errors" + "io" + "maps" + "slices" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/gopherium/framework/gonsole/testkit" + + "github.com/gopherium/alphone/internal/contact" + "github.com/gopherium/alphone/internal/postgres" + "github.com/gopherium/alphone/sdk" +) + +// identityReadDeadline is how long a wiring test waits for an identity the feed still holds back. +const identityReadDeadline = 30 * time.Second + +// identityReadPause is how long a wiring test waits between two reads of the feed. +const identityReadPause = 20 * time.Millisecond + +// readingSeams returns plugins that read the seams run hands them through read, then stop the run. +func readingSeams(read func(sdk.Deps)) func(sdk.Deps) ([]sdk.Plugin, error) { + return func(deps sdk.Deps) ([]sdk.Plugin, error) { + read(deps) + return nil, errCaptured + } +} + +// runReading runs the server over env with plugins that read the seams through read, failing unless the run stopped. +func runReading(t *testing.T, env map[string]string, read func(sdk.Deps)) { + t.Helper() + if err := run(t.Context(), testkit.Getenv(env), io.Discard, readingSeams(read)); !errors.Is(err, errCaptured) { + t.Fatalf("run() error = %v, want %v in its chain", err, errCaptured) + } +} + +// standingInNewTenant stores a tenant in the database behind pool and returns a context serving it. +func standingInNewTenant(t *testing.T, pool *pgxpool.Pool) context.Context { + t.Helper() + id := uuid.Must(uuid.NewV7()) + if _, err := pool.Exec(t.Context(), "INSERT INTO core.tenants (id, name) VALUES ($1, $2)", id, "Acme"); err != nil { + t.Fatalf("seeding the tenant: %v", err) + } + return sdk.WithTenant(t.Context(), id) +} + +// emailOwner is a contact's name beside every email address it owns. +type emailOwner struct { + name string + emails []string +} + +// storedEmailOwner stores the contact owning every email address in the tenant ctx serves and returns its id. +func storedEmailOwner(t *testing.T, pool *pgxpool.Pool, ctx context.Context, owned emailOwner) uuid.UUID { + t.Helper() + addresses := make([]contact.Address, 0, len(owned.emails)) + for _, email := range owned.emails { + addresses = append(addresses, contact.Address{Channel: "email", Identifier: email}) + } + owner, _, err := contact.NewResolver(postgres.NewContactStore(pool)).CreateWithIdentities(ctx, owned.name, addresses) + if err != nil { + t.Fatalf("storing %s: %v", owned.name, err) + } + return owner.ID +} + +// settledIdentities reads the email feed past the cursor until it answers want identities or the deadline passes. +func settledIdentities( + ctx context.Context, feed sdk.IdentityFeed, after sdk.IdentityCursor, want int, +) ([]sdk.AddedIdentity, sdk.IdentityCursor, error) { + deadline := time.Now().Add(identityReadDeadline) + for { + answered, next, err := feed.IdentitiesAddedSince(ctx, "email", after, want+1) + if err != nil { + return nil, after, err + } + if len(answered) >= want || time.Now().After(deadline) { + return answered, next, nil + } + time.Sleep(identityReadPause) + } +} + +// feedReads are the pages a wiring test reads from the email feed. +type feedReads struct { + first, rest, past []sdk.AddedIdentity +} + +// readFeedPages reads a settled first page from the start, the page after it, then a page from the horizon. +func readFeedPages(ctx context.Context, feed sdk.IdentityFeed) (feedReads, error) { + var reads feedReads + first, cursor, err := settledIdentities(ctx, feed, sdk.IdentityCursor{}, 2) + if err != nil { + return reads, err + } + reads.first = first + if reads.rest, _, err = feed.IdentitiesAddedSince(ctx, "email", cursor, 10); err != nil { + return reads, err + } + horizon, err := feed.Horizon(ctx) + if err != nil { + return reads, err + } + reads.past, _, err = feed.IdentitiesAddedSince(ctx, "email", horizon, 10) + return reads, err +} + +// wantIdentitiesOf fails the test unless answered holds exactly the want identifiers, each of owner and timed. +func wantIdentitiesOf(t *testing.T, answered []sdk.AddedIdentity, owner uuid.UUID, want []string) { + t.Helper() + var got []string + for _, added := range answered { + if added.ContactID != owner || added.AddedAt.IsZero() { + t.Errorf("the feed answered %+v, want an identity of %s carrying the time it was added", added, owner) + } + got = append(got, added.Identifier) + } + if slices.Sort(got); !slices.Equal(got, want) { + t.Errorf("the feed of the caller's tenant answered %v, want %v", got, want) + } +} + +func TestTheContactMatchBoundFallsBackToItsDefault(t *testing.T) { + t.Parallel() + + held, err := loadRunConfig(testkit.Getenv(map[string]string{"ALPHONE_DATABASE_URL": unreachableDatabaseURL})) + + if err != nil { + t.Fatalf("loadRunConfig() error = %v, want nil", err) + } + if held.contactMatchMax != contact.DefaultMatchMax { + t.Errorf("contactMatchMax = %d, want the default %d", held.contactMatchMax, contact.DefaultMatchMax) + } +} + +func TestRunHandsPluginsAMatcherOverTheirOwnTenant(t *testing.T) { + t.Parallel() + + databaseURL := testDatabaseURL(t) + pool := testPool(t, databaseURL) + acme := standingInNewTenant(t, pool) + mine := storedEmailOwner(t, pool, acme, emailOwner{"Maria Perez", []string{"maria@example.com"}}) + storedEmailOwner(t, pool, t.Context(), emailOwner{"Ada Lovelace", []string{"ada@example.com"}}) + var matched map[string]sdk.Contact + var matchErr error + + runReading(t, map[string]string{"ALPHONE_DATABASE_URL": databaseURL}, func(deps sdk.Deps) { + matched, matchErr = deps.Matcher.MatchIdentities(acme, "email", []string{"Maria@Example.com", "ada@example.com"}) + }) + + if matchErr != nil { + t.Fatalf("MatchIdentities() error = %v, want nil", matchErr) + } + want := map[string]sdk.Contact{"maria@example.com": {ID: mine, Name: "Maria Perez"}} + if !maps.Equal(matched, want) { + t.Errorf("MatchIdentities() = %+v, want only Maria Perez of the caller's own tenant", matched) + } +} + +func TestRunHandsPluginsAMatcherHeldToTheConfiguredBound(t *testing.T) { + t.Parallel() + + var matchErr error + + runReading(t, map[string]string{ + "ALPHONE_DATABASE_URL": unreachableDatabaseURL, + "ALPHONE_CONTACT_MATCH_MAX": "2", + }, func(deps sdk.Deps) { + _, matchErr = deps.Matcher.MatchIdentities(t.Context(), "email", + []string{"maria@example.com", "ada@example.com", "grace@example.com"}) + }) + + if !errors.Is(matchErr, sdk.ErrTooManyIdentifiers) { + t.Errorf("MatchIdentities() of three error = %v, want %v under a bound of two", matchErr, sdk.ErrTooManyIdentifiers) + } +} + +func TestRunHandsPluginsAFeedOverTheirOwnTenantInPagesHeldToTheBound(t *testing.T) { + t.Parallel() + + databaseURL := testDatabaseURL(t) + pool := testPool(t, databaseURL) + acme := standingInNewTenant(t, pool) + want := []string{"maria.perez@example.com", "maria@example.com", "mperez@example.com"} + mine := storedEmailOwner(t, pool, acme, emailOwner{"Maria Perez", want}) + storedEmailOwner(t, pool, t.Context(), emailOwner{"Ada Lovelace", []string{"ada@example.com"}}) + var reads feedReads + var readErr error + + runReading(t, map[string]string{ + "ALPHONE_DATABASE_URL": databaseURL, + "ALPHONE_CONTACT_MATCH_MAX": "2", + }, func(deps sdk.Deps) { + reads, readErr = readFeedPages(acme, deps.Identities) + }) + + if readErr != nil { + t.Fatalf("reading the feed error = %v, want nil", readErr) + } + if len(reads.first) != 2 { + t.Errorf("the first page answered %d identities, want the bound of 2", len(reads.first)) + } + wantIdentitiesOf(t, slices.Concat(reads.first, reads.rest), mine, want) + if len(reads.past) != 0 { + t.Errorf("a reader starting at the horizon answered %+v, want none of the identities it settled", reads.past) + } +} diff --git a/cmd/alphone/loadplugins_test.go b/cmd/alphone/loadplugins_test.go index 24e4de3b..008a3f70 100644 --- a/cmd/alphone/loadplugins_test.go +++ b/cmd/alphone/loadplugins_test.go @@ -93,6 +93,10 @@ func TestLoadPluginsNamesWhatItCannotRead(t *testing.T) { map[string]string{"ALPHONE_DATABASE_URL": unreachableDatabaseURL, "ALPHONE_SHUTDOWN_STOP_GRACE": "0s"}, `ALPHONE_SHUTDOWN_STOP_GRACE: must stand above zero, got "0s"`, }, + "a contact match bound of zero": { + map[string]string{"ALPHONE_DATABASE_URL": unreachableDatabaseURL, "ALPHONE_CONTACT_MATCH_MAX": "0"}, + `ALPHONE_CONTACT_MATCH_MAX: must stand above zero, got "0"`, + }, "a mail port in words": { map[string]string{ "ALPHONE_DATABASE_URL": unreachableDatabaseURL, diff --git a/cmd/alphone/plugins.go b/cmd/alphone/plugins.go index 1087624a..1fd8faa2 100644 --- a/cmd/alphone/plugins.go +++ b/cmd/alphone/plugins.go @@ -134,3 +134,61 @@ func (b directoryBridge) CreateWithIdentities( } return sdk.Contact{ID: owner.ID, Name: owner.Name}, created, nil } + +type matcherBridge struct { + matcher *contact.Matcher +} + +// MatchIdentities returns the [sdk.Contact] owning each identifier one owns, keyed by its canonical form. +func (b matcherBridge) MatchIdentities( + ctx context.Context, channel sdk.Channel, identifiers []string, +) (map[string]sdk.Contact, error) { + owners, err := b.matcher.MatchIdentities(ctx, contact.Channel(channel), identifiers) + if errors.Is(err, contact.ErrTooManyIdentifiers) { + return nil, fmt.Errorf("%w: %w", sdk.ErrTooManyIdentifiers, err) + } + if err != nil { + return nil, markInvalid(err) + } + matched := make(map[string]sdk.Contact, len(owners)) + for identifier, owner := range owners { + matched[identifier] = sdk.Contact{ID: owner.ID, Name: owner.Name} + } + return matched, nil +} + +type identityFeedBridge struct { + feed *contact.Feed +} + +// IdentitiesAddedSince returns at most limit [sdk.AddedIdentity] values past the cursor and the next cursor. +func (b identityFeedBridge) IdentitiesAddedSince( + ctx context.Context, channel sdk.Channel, cursor sdk.IdentityCursor, limit int, +) ([]sdk.AddedIdentity, sdk.IdentityCursor, error) { + added, next, err := b.feed.IdentitiesAddedSince(ctx, contact.Channel(channel), contact.Cursor(cursor), limit) + if err != nil { + return nil, cursor, markFeedRefusal(err) + } + answered := make([]sdk.AddedIdentity, len(added)) + for i, identity := range added { + answered[i] = sdk.AddedIdentity(identity) + } + return answered, sdk.IdentityCursor(next), nil +} + +// markFeedRefusal returns err with a refused feed read marked for the plugin. +func markFeedRefusal(err error) error { + switch { + case errors.Is(err, contact.ErrCursorStale): + return fmt.Errorf("%w: %w", sdk.ErrCursorStale, err) + case errors.Is(err, contact.ErrInvalidLimit): + return fmt.Errorf("%w: %w", sdk.ErrInvalidLimit, err) + } + return markInvalid(err) +} + +// Horizon returns the cursor a new reader starts from. +func (b identityFeedBridge) Horizon(ctx context.Context) (sdk.IdentityCursor, error) { + horizon, err := b.feed.Horizon(ctx) + return sdk.IdentityCursor(horizon), err +} diff --git a/cmd/alphone/plugins_test.go b/cmd/alphone/plugins_test.go index b7a088a5..d654db70 100644 --- a/cmd/alphone/plugins_test.go +++ b/cmd/alphone/plugins_test.go @@ -38,11 +38,187 @@ func (failingStore) CreateContactWithIdentities(context.Context, contact.Contact return errStore } +// MatchIdentities returns no contact and errStore. +func (failingStore) MatchIdentities(context.Context, contact.Channel, []string) (map[string]contact.Contact, error) { + return nil, errStore +} + +// IdentitiesAddedSince returns no identity, the cursor it was given and errStore. +func (failingStore) IdentitiesAddedSince( + _ context.Context, _ contact.Channel, after contact.Cursor, _ int, +) ([]contact.AddedIdentity, contact.Cursor, error) { + return nil, after, errStore +} + +// IdentityHorizon returns the zero cursor and errStore. +func (failingStore) IdentityHorizon(context.Context) (contact.Cursor, error) { + return contact.Cursor{}, errStore +} + +// IdentityCursorCheck returns an empty check and errStore. +func (failingStore) IdentityCursorCheck(context.Context, contact.Cursor) (contact.CursorCheck, error) { + return contact.CursorCheck{}, errStore +} + // newDirectoryBridge returns a directory bridge whose store always fails. func newDirectoryBridge() directoryBridge { return directoryBridge{resolver: contact.NewResolver(failingStore{})} } +// newMatcherBridge returns a matcher bridge answering two identifiers a call over a store that always fails. +func newMatcherBridge() matcherBridge { + return matcherBridge{matcher: contact.NewMatcher(failingStore{}, 2)} +} + +// newIdentityFeedBridge returns a feed bridge answering two identities a call over a store that always fails. +func newIdentityFeedBridge() identityFeedBridge { + return identityFeedBridge{feed: contact.NewFeed(failingStore{}, 2)} +} + +// someCursor returns a cursor past the start of the feed. +func someCursor() sdk.IdentityCursor { + return sdk.IdentityCursor{Server: 184467235, Transaction: 7, Identity: uuid.Must(uuid.NewV7())} +} + +func TestMatcherBridgeMarksTheRefusals(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + channel sdk.Channel + identifiers []string + want error + }{ + "a blank channel": {" ", []string{"maria@example.com"}, sdk.ErrInvalidContact}, + "a phone holding no digits": {"phone", []string{"n/a"}, sdk.ErrInvalidContact}, + "a batch above the bound": { + "email", []string{"maria@example.com", "ada@example.com", "grace@example.com"}, sdk.ErrTooManyIdentifiers, + }, + } + + for name, tt := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + matched, err := newMatcherBridge().MatchIdentities(t.Context(), tt.channel, tt.identifiers) + + if !errors.Is(err, tt.want) { + t.Errorf("MatchIdentities() error = %v, want a plugin to read it as %v", err, tt.want) + } + if matched != nil { + t.Errorf("MatchIdentities() = %+v on a refusal, want nothing", matched) + } + }) + } +} + +func TestMatcherBridgeLeavesAStoreFailureUnmarked(t *testing.T) { + t.Parallel() + + _, err := newMatcherBridge().MatchIdentities(t.Context(), "email", []string{"maria@example.com"}) + + if !errors.Is(err, errStore) { + t.Fatalf("MatchIdentities() error = %v, want %v", err, errStore) + } + if errors.Is(err, sdk.ErrInvalidContact) || errors.Is(err, sdk.ErrTooManyIdentifiers) { + t.Errorf("a store failure was marked as a refusal %v, so a plugin would drop the batch rather than retry", err) + } +} + +func TestIdentityFeedBridgeMarksABlankChannelAndKeepsTheCursor(t *testing.T) { + t.Parallel() + + cursor := someCursor() + + added, next, err := newIdentityFeedBridge().IdentitiesAddedSince(t.Context(), " ", cursor, 1) + + if !errors.Is(err, sdk.ErrInvalidContact) { + t.Errorf("IdentitiesAddedSince() error = %v, want a plugin to read it as %v", err, sdk.ErrInvalidContact) + } + if added != nil || next != cursor { + t.Errorf("IdentitiesAddedSince() = %+v at %+v on a refusal, want nothing at %+v", added, next, cursor) + } +} + +func TestIdentityFeedBridgeMarksALimitBelowOneAndKeepsTheCursor(t *testing.T) { + t.Parallel() + + cursor := someCursor() + + added, next, err := newIdentityFeedBridge().IdentitiesAddedSince(t.Context(), "email", cursor, 0) + + if !errors.Is(err, sdk.ErrInvalidLimit) || !errors.Is(err, contact.ErrInvalidLimit) { + t.Errorf("IdentitiesAddedSince() error = %v, want a plugin to read it as %v", err, sdk.ErrInvalidLimit) + } + if added != nil || next != cursor { + t.Errorf("IdentitiesAddedSince() = %+v at %+v on a refusal, want nothing at %+v", added, next, cursor) + } +} + +// checkedFeedStore answers an empty page for every cursor and the fixed check of each one. +type checkedFeedStore struct { + check contact.CursorCheck +} + +// IdentitiesAddedSince returns no identity and the cursor it was given. +func (checkedFeedStore) IdentitiesAddedSince( + _ context.Context, _ contact.Channel, after contact.Cursor, _ int, +) ([]contact.AddedIdentity, contact.Cursor, error) { + return nil, after, nil +} + +// IdentityHorizon returns the zero cursor. +func (checkedFeedStore) IdentityHorizon(context.Context) (contact.Cursor, error) { + return contact.Cursor{}, nil +} + +// IdentityCursorCheck returns the fixed check. +func (s checkedFeedStore) IdentityCursorCheck(context.Context, contact.Cursor) (contact.CursorCheck, error) { + return s.check, nil +} + +func TestIdentityFeedBridgeMarksAStaleCursorAndKeepsIt(t *testing.T) { + t.Parallel() + + cursor := someCursor() + tests := map[string]contact.CursorCheck{ + "a cursor past the server": {Server: cursor.Server, Ahead: true}, + "a cursor another server numbered": {Server: cursor.Server + 1}, + } + + for name, check := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + bridge := identityFeedBridge{feed: contact.NewFeed(checkedFeedStore{check: check}, 2)} + + added, next, err := bridge.IdentitiesAddedSince(t.Context(), "email", cursor, 1) + + if !errors.Is(err, sdk.ErrCursorStale) || !errors.Is(err, contact.ErrCursorStale) { + t.Errorf("IdentitiesAddedSince() error = %v, want a plugin to read it as %v", err, sdk.ErrCursorStale) + } + if added != nil || next != cursor { + t.Errorf("IdentitiesAddedSince() = %+v at %+v on a refusal, want nothing at %+v", added, next, cursor) + } + }) + } +} + +func TestIdentityFeedBridgeLeavesAStoreFailureUnmarked(t *testing.T) { + t.Parallel() + + bridge, cursor := newIdentityFeedBridge(), someCursor() + + _, next, err := bridge.IdentitiesAddedSince(t.Context(), "email", cursor, 1) + if !errors.Is(err, errStore) || errors.Is(err, sdk.ErrInvalidContact) { + t.Errorf("IdentitiesAddedSince() error = %v, want %v left unmarked", err, errStore) + } + if next != cursor { + t.Errorf("IdentitiesAddedSince() moved the cursor to %+v on a failure, want it kept at %+v", next, cursor) + } + if _, err := bridge.Horizon(t.Context()); !errors.Is(err, errStore) { + t.Errorf("Horizon() error = %v, want %v", err, errStore) + } +} + func TestDirectoryBridgeMarksUnusableDetails(t *testing.T) { t.Parallel() diff --git a/cmd/alphone/run.go b/cmd/alphone/run.go index d84b9e4a..9f58b4b2 100644 --- a/cmd/alphone/run.go +++ b/cmd/alphone/run.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "log/slog" + "math" "net/http" "net/url" "os" @@ -21,6 +22,7 @@ import ( "github.com/gopherium/gouncer/authkit/ratelimit" "github.com/gopherium/alphone/graph" + "github.com/gopherium/alphone/internal/contact" "github.com/gopherium/alphone/internal/graphres" "github.com/gopherium/alphone/internal/graphroot" "github.com/gopherium/alphone/internal/postgres" @@ -295,11 +297,12 @@ type runConfig struct { // composeSettings carries the settings compose hands the plugins and the stores it builds. type composeSettings struct { - machineGrace time.Duration - tenants tenantSettings - mail mailSettings - webhookHosts webhook.AllowList - paging graphres.Paging + machineGrace time.Duration + tenants tenantSettings + contactMatchMax int + mail mailSettings + webhookHosts webhook.AllowList + paging graphres.Paging } // servingDefaults are the HTTP timeouts and shutdown graces the server runs under when the environment names none. @@ -473,6 +476,10 @@ func loadComposeSettings(env gonsole.Env) (composeSettings, error) { if err != nil { return composeSettings{}, err } + contactMatchMax, err := env.Count("CONTACT_MATCH_MAX", contact.DefaultMatchMax, gonsole.AtMost(math.MaxInt32)) + if err != nil { + return composeSettings{}, err + } mail, err := loadMailSettings(env) if err != nil { return composeSettings{}, err @@ -486,7 +493,8 @@ func loadComposeSettings(env gonsole.Env) (composeSettings, error) { return composeSettings{}, err } return composeSettings{ - machineGrace: machineGrace, tenants: tenants, mail: mail, webhookHosts: webhookHosts, paging: paging, + machineGrace: machineGrace, tenants: tenants, contactMatchMax: contactMatchMax, + mail: mail, webhookHosts: webhookHosts, paging: paging, }, nil } diff --git a/cmd/alphone/settings_test.go b/cmd/alphone/settings_test.go index 9052be73..d601db33 100644 --- a/cmd/alphone/settings_test.go +++ b/cmd/alphone/settings_test.go @@ -34,6 +34,7 @@ func coreSettings() map[string]string { "ALPHONE_TENANT_MACHINE_GRACE": "72h", "ALPHONE_TENANTS_HELD": "8", "ALPHONE_TENANTS_REFRESH": "30s", + "ALPHONE_CONTACT_MATCH_MAX": "40", "ALPHONE_SMTP_HOST": "mail.example.com", "ALPHONE_SMTP_PORT": "2525", "ALPHONE_SMTP_USERNAME": "crm", @@ -636,6 +637,11 @@ func TestSettingRefusalsNameTheSettingAndTheReason(t *testing.T) { "ALPHONE_WEBHOOK_ALLOWED_HOSTS", "127.0.0.1/32,n8n", "must list CIDR ranges such as 127.0.0.1/32 or host names with a port such as n8n:5678", }, + "a match bound in words": {"ALPHONE_CONTACT_MATCH_MAX", "a hundred", "must be a whole number"}, + "a match of no identifier": {"ALPHONE_CONTACT_MATCH_MAX", "0", "must stand above zero"}, + "a match past a query row bound": { + "ALPHONE_CONTACT_MATCH_MAX", "2147483648", "must stand at or below 2147483647", + }, } for testName, tt := range tests { t.Run(testName, func(t *testing.T) { diff --git a/docs/astro.config.mjs b/docs/astro.config.mjs index 935d1356..edcf347d 100644 --- a/docs/astro.config.mjs +++ b/docs/astro.config.mjs @@ -65,6 +65,7 @@ export default defineConfig({ { slug: 'extending/graph' }, { slug: 'extending/commands' }, { slug: 'extending/events' }, + { slug: 'extending/contacts' }, ], }, { diff --git a/docs/src/content/docs/extending/contacts.md b/docs/src/content/docs/extending/contacts.md new file mode 100644 index 00000000..c17a3b98 --- /dev/null +++ b/docs/src/content/docs/extending/contacts.md @@ -0,0 +1,218 @@ +--- +title: Contacts +description: Find the contacts behind the addresses your plugin meets, one or many at a time, and follow the addresses contacts gain. +--- + +A channel plugin, such as WhatsApp or a mail plugin, meets people by their +address: a phone number, a WhatsApp number or an email address. AlphOne keeps +one contact per person and links each of their addresses to it. The SDK calls +a linked address an **identity**. It is made of a channel, such as `email`, +and an identifier, such as `maria.perez@example.com`. + +Your plugin gets four helpers for contacts in `sdk.Deps`. Keep the ones you +need from `Register`: + +| Field | What it does | +| --- | --- | +| `deps.Resolver` | Finds the contact behind one address, and creates one when no contact has it. | +| `deps.Contacts` | Looks one address up without creating anything, and creates a contact that owns several addresses. | +| `deps.Matcher` | Finds the contacts behind many addresses in one call. | +| `deps.Identities` | Lists the addresses contacts gain, in a fixed order. | + +## Every call stays in one workspace + +Every helper works in the workspace `ctx` carries, and never sees another +workspace's contacts. AlphOne puts the workspace on `ctx` for every signed in +request. Work your plugin starts on its own, such as a background sweep, sets +it with `sdk.WithTenant`, the way the WhatsApp media fetcher does. A `ctx` that +carries no workspace works in the default one. + +## How addresses are compared + +Every helper puts an address in its channel's canonical form before it +compares or stores it: + +- The channel is trimmed and lower-cased, so `Email` is `email`. +- An email address is trimmed and lower-cased. `Maria.Perez@Example.com` is + the same address as `maria.perez@example.com`. Nothing else is folded, so + `maria.perez+news@example.com` and `mariaperez@example.com` are other + addresses. +- A phone number keeps only its digits, with a leading `+` when it starts + with one. `+184 467 235` becomes `+184467235`. +- On any other channel, such as `whatsapp`, the identifier is trimmed and + then compared exactly. + +## Find or create one contact + +`deps.Resolver` is a `sdk.ContactResolver`. Its `Resolve` answers the contact +that owns an address. When no contact owns it, `Resolve` creates one and links +the address to it. The new contact is named after the display name you pass, +or after the address when the name is blank. Two calls racing on one new +address get the same contact. The WhatsApp plugin calls it for every message +that arrives, so each new sender becomes a contact: + +```go +owner, err := p.resolver.Resolve(ctx, "whatsapp", m.sender, m.senderName) +if err != nil { + return err +} +``` + +`deps.Contacts` is a `sdk.ContactDirectory`, for when an unknown address must +not become a contact on its own. `FindByIdentity` answers the contact that +owns one address, and whether one does. `CreateWithIdentities` creates a +contact that owns several addresses at once, as the importer does for each +row. When another contact already owns one of the addresses, it creates +nothing, answers that contact and reports `false`. + +## Match many addresses at once + +`deps.Matcher` is a `sdk.ContactMatcher`. Its `MatchIdentities` takes a list +of addresses on one channel and answers, in one database read, the contact +behind each address a contact owns. It never creates a contact, and an address +no contact owns is left out of the answer. Use it where one message names many +people, such as the sender and every recipient of a mail. + +The answer is a map keyed by each address in its canonical form, so look an +address up in that form: + +```go +owners, err := p.matcher.MatchIdentities(ctx, "email", []string{ + "Maria.Perez@Example.com", + "ada.lovelace@example.com", +}) +if err != nil { + return err +} +owner, found := owners["maria.perez@example.com"] +``` + +One call takes at most `deps.ContactMatchMax` addresses. The server sets it +with `ALPHONE_CONTACT_MATCH_MAX`, 100 when unset, see +[Configuration](/self-hosting/configuration/#contacts). Every address in the +list counts, a repeated one too. A longer list is refused whole with +`sdk.ErrTooManyIdentifiers`, and nothing is matched, so split a long list into +batches of `deps.ContactMatchMax`. + +One address that is empty in its canonical form, such as a blank one or a +phone number holding no digits, also refuses the whole list, with +`sdk.ErrInvalidContact`. Drop such addresses before the call. + +## Follow the addresses contacts gain + +`deps.Identities` is a `sdk.IdentityFeed`. It lists every address the +workspace's contacts gain on one channel, however it was added: with a new +contact, on a contact's page, by an import or by a plugin. Use it when your +plugin keeps data only for known contacts and must catch up when a contact +appears or gains an address. A mail plugin, for example, fetches the recent +mail of each new address. + +Your plugin keeps a `sdk.IdentityCursor`, which marks how far it has read. +Each call answers the next page and the cursor after it. This loop asks for +pages of `deps.ContactMatchMax`, kept in `p.matchMax`, so each page fits one +match. It starts again from the zero cursor when the feed refuses a stale +one: + +```go +for { + added, next, err := p.identities.IdentitiesAddedSince( + ctx, "email", cursor, p.matchMax, + ) + if errors.Is(err, sdk.ErrCursorStale) { + cursor = sdk.IdentityCursor{} + continue + } + if err != nil { + return err + } + if len(added) == 0 { + return nil + } + for _, identity := range added { + err = p.fetchRecentMail(ctx, identity) + if err != nil { + return err + } + } + cursor = next + err = p.saveCursor(ctx, cursor) + if err != nil { + return err + } +} +``` + +Each `sdk.AddedIdentity` holds the `ContactID` that gained the address, the +`Identifier` in its canonical form, and `AddedAt`, the time AlphOne saved it. + +A few rules keep a reader correct: + +- A page holds at most the `limit` you pass, and never more than + `deps.ContactMatchMax`, so a page always fits one call to the matcher. + A page can hold fewer than you asked while more wait, so read until a page + comes back empty. An empty page answers the cursor you passed. A `limit` + below one is refused with `sdk.ErrInvalidLimit`. +- Save the cursor only after you have handled its page. A plugin that stops + halfway starts again from its saved cursor and handles part of that page a + second time, so handling an address twice must do no harm. +- Keep one cursor for each workspace and channel you follow. +- The zero `sdk.IdentityCursor{}` is the place before every address. A reader + starting there lists every address the workspace's contacts hold today, and + then each new one. `Horizon` answers the place a new reader starts from to + see only what comes next: every address whose save had not finished when + you asked, and every later one. Addresses saved since the oldest write + still running on the PostgreSQL server began can show up too, so a long + open transaction can make that a long list. +- Each cursor remembers the PostgreSQL server that answered it. Save all + three of its fields: `Server`, `Transaction` and `Identity`. Keep the saved + cursor in AlphOne's own database, the one at `deps.DatabaseURL`, so a + restore brings it back beside the addresses. +- The feed refuses a cursor from another server with `sdk.ErrCursorStale`. + That happens after a dump is restored onto a new server, such as one for a + new PostgreSQL major version. A cursor past the server's count is refused + the same way. Start again from the zero cursor and handle every address + once more. +- The feed reports only additions. A removed address is not reported, and one + removed before you read it never shows up. +- Follow the cursor, never `AddedAt`. AlphOne stamps that time as it saves, + and a save can finish after a later one. + +## Why the feed reads by transaction + +PostgreSQL numbers every transaction that writes, counting up, and the feed +orders addresses by the number of the transaction that saved them. It answers +only addresses whose transaction number is lower than that of every +transaction still running. So a save that started first but finishes last is +never skipped: the feed holds every later address back until it finishes. The +price is that a long open transaction makes the feed wait rather than skip. +Any open transaction that writes counts, in any database on the same +PostgreSQL server, not only AlphOne's, and the feed answers nothing past it +until it ends. + +Transaction numbers belong to one PostgreSQL server. A dump restored onto +another server, such as a new PostgreSQL major version, keeps the numbers +the old server gave, while the new server counts from the start again. The +two servers can hand out the same numbers, so a number alone cannot say +where a reader stopped. That is why each cursor also names its server. +PostgreSQL gives a server its identity when the server is first set up, and a +copy of its data files, such as a promoted replica, keeps that identity. So +restore a dump onto a server set up fresh, never onto a copy of the server +the dump came from. The feed refuses a cursor from another server, or one +past the server's count, with `sdk.ErrCursorStale`, as the rules above say. +Each time AlphOne starts, +it renumbers every address another server numbered, and every one numbered +past the server's count, as saved by that start. So a reader from the zero +cursor sees them at once. Say the database moves to another server and back. +A reader whose cursor this server answered before the move still gets the +addresses saved on the other server. + +## When a call is refused + +| Error | Cause | +| --- | --- | +| `sdk.ErrInvalidContact` | The channel is empty, an address is empty in its canonical form, or a new contact's name is blank | +| `sdk.ErrTooManyIdentifiers` | A match asked for more addresses than `deps.ContactMatchMax` | +| `sdk.ErrInvalidLimit` | A feed page asked for fewer than one address | +| `sdk.ErrCursorStale` | A feed cursor comes from another PostgreSQL server, or names a transaction the server has not reached | + +Test each with `errors.Is`. Any other error comes from the database. diff --git a/docs/src/content/docs/self-hosting/configuration.md b/docs/src/content/docs/self-hosting/configuration.md index 14f7fe52..76d47099 100644 --- a/docs/src/content/docs/self-hosting/configuration.md +++ b/docs/src/content/docs/self-hosting/configuration.md @@ -18,8 +18,8 @@ When a variable holds a value AlphOne refuses: [Account records and tokens](#account-records-and-tokens), which only the commands read. - A refused value under [Mail](#mail), [Workspaces](#workspaces), - [Webhooks](#webhooks), [Fields plugin](#fields-plugin) or - [WhatsApp plugin](#whatsapp-plugin), + [Contacts](#contacts), [Webhooks](#webhooks), + [Fields plugin](#fields-plugin) or [WhatsApp plugin](#whatsapp-plugin), in `ALPHONE_GRAPH_PAGE_SIZE` or `ALPHONE_GRAPH_PAGE_CAP`, or in `ALPHONE_SHUTDOWN_STOP_GRACE`, also stops the account commands that take `-as`, before they change anything. It stops `migrate`, @@ -93,6 +93,15 @@ deployment serves more than one. | `ALPHONE_TENANTS_HELD` | How many workspaces' fields each server process keeps in memory. Defaults to 256. Must be a whole number above zero. | | `ALPHONE_TENANTS_REFRESH` | How long a server process keeps a workspace's fields before reading them again. Defaults to `1m`. Must be a duration above zero. | +## Contacts + +Plugins look up the contacts behind many addresses at once, such as every +recipient of a mail, see [Contacts](/extending/contacts/). + +| Variable | Purpose | +| --- | --- | +| `ALPHONE_CONTACT_MATCH_MAX` | The most addresses a plugin looks up in one call. It also caps one page of the addresses contacts gain, so a page always fits one lookup. Defaults to 100. A value that is not a whole number between 1 and 2147483647 stops `serve` and the other commands listed at the top of this page. | + ## Lists, toasts and formats | Variable | Purpose | diff --git a/docs/src/content/docs/self-hosting/updates-and-backups.md b/docs/src/content/docs/self-hosting/updates-and-backups.md index 7628502f..b251e0f8 100644 --- a/docs/src/content/docs/self-hosting/updates-and-backups.md +++ b/docs/src/content/docs/self-hosting/updates-and-backups.md @@ -263,3 +263,10 @@ gunzip -c backups/alphone-.sql.gz | \ docker compose exec -T postgres psql -U alphone alphone docker compose start alphone ``` + +A dump can also go onto a new PostgreSQL server, such as a fresh container +for a new PostgreSQL major version. The new server numbers its transactions +from the start again. A plugin that follows the addresses contacts gain, see +[Contacts](/extending/contacts/), saved a cursor that names the old server. +The feed refuses that cursor, so the plugin reads every address again from +the beginning. Handling an address twice does no harm. diff --git a/internal/contact/feed.go b/internal/contact/feed.go new file mode 100644 index 00000000..f91b2254 --- /dev/null +++ b/internal/contact/feed.go @@ -0,0 +1,107 @@ +// SPDX-License-Identifier: Elastic-2.0 + +package contact + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/google/uuid" +) + +// ErrInvalidLimit reports a page asked to hold fewer than one identity. +var ErrInvalidLimit = errors.New("contact: invalid limit") + +// ErrCursorStale reports a cursor this database server cannot continue. +var ErrCursorStale = errors.New("contact: cursor stale") + +// Cursor marks a place in the order identities were added, its zero value the place before every identity. +type Cursor struct { + // Server identifies the PostgreSQL server that numbered the transaction, zero in the zero cursor. + Server int64 + // Transaction is the transaction that added the identity at the place. + Transaction uint64 + // Identity is the id of the identity at the place. + Identity uuid.UUID +} + +// AddedIdentity is one identity the feed answers. +type AddedIdentity struct { + ContactID uuid.UUID + Identifier string + AddedAt time.Time +} + +// CursorCheck is what the database server answers about a cursor. +type CursorCheck struct { + // Server identifies the PostgreSQL server that answered. + Server int64 + // Ahead reports whether the cursor names a transaction past every one the server has begun. + Ahead bool +} + +// FeedStore lists identities in the order they were added. +type FeedStore interface { + IdentitiesAddedSince( + ctx context.Context, channel Channel, after Cursor, limit int, + ) ([]AddedIdentity, Cursor, error) + IdentityHorizon(ctx context.Context) (Cursor, error) + IdentityCursorCheck(ctx context.Context, cursor Cursor) (CursorCheck, error) +} + +// Feed lists the identities each channel gains, in an order no late commit can skip. +type Feed struct { + store FeedStore + bound int +} + +// NewFeed returns a [Feed] over store answering at most bound identities a call. +func NewFeed(store FeedStore, bound int) *Feed { + return &Feed{store: store, bound: bound} +} + +// IdentitiesAddedSince returns at most limit identities past the cursor, within the bound, and the next cursor. +func (f *Feed) IdentitiesAddedSince( + ctx context.Context, channel Channel, after Cursor, limit int, +) ([]AddedIdentity, Cursor, error) { + normalizedChannel := normalizeChannel(channel) + if normalizedChannel == "" { + return nil, after, ErrEmptyChannel + } + if limit < 1 { + return nil, after, ErrInvalidLimit + } + server, err := f.checkCursor(ctx, after) + if err != nil { + return nil, after, err + } + added, next, err := f.store.IdentitiesAddedSince(ctx, normalizedChannel, after, min(limit, f.bound)) + if err != nil || len(added) == 0 { + return nil, after, err + } + next.Server = server + return added, next, nil +} + +// checkCursor returns the server's identity, or [ErrCursorStale] when the server cannot continue the cursor. +func (f *Feed) checkCursor(ctx context.Context, after Cursor) (int64, error) { + check, err := f.store.IdentityCursorCheck(ctx, after) + if err != nil { + return 0, err + } + if after != (Cursor{}) && after.Server != check.Server { + return 0, fmt.Errorf("%w: the cursor names the server %d, not this server %d", + ErrCursorStale, after.Server, check.Server) + } + if check.Ahead { + return 0, fmt.Errorf("%w: transaction %d is past the server's count", ErrCursorStale, after.Transaction) + } + return check.Server, nil +} + +// Horizon returns the cursor a new reader starts from. +func (f *Feed) Horizon(ctx context.Context) (Cursor, error) { + return f.store.IdentityHorizon(ctx) +} diff --git a/internal/contact/feed_test.go b/internal/contact/feed_test.go new file mode 100644 index 00000000..8a2ae98b --- /dev/null +++ b/internal/contact/feed_test.go @@ -0,0 +1,233 @@ +// SPDX-License-Identifier: Elastic-2.0 + +package contact_test + +import ( + "errors" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + "github.com/google/uuid" + + "github.com/gopherium/alphone/internal/contact" +) + +func TestIdentitiesAddedSinceAsksForTheCanonicalChannel(t *testing.T) { + t.Parallel() + + after := contact.Cursor{Transaction: 7, Identity: uuid.Must(uuid.NewV7())} + index := &recordingIndex{ + added: []contact.AddedIdentity{{ + ContactID: uuid.Must(uuid.NewV7()), Identifier: "maria@example.com", AddedAt: time.Now(), + }}, + next: contact.Cursor{Transaction: 9, Identity: uuid.Must(uuid.NewV7())}, + } + + got, next, err := contact.NewFeed(index, 5).IdentitiesAddedSince(t.Context(), " Email ", after, 5) + + if err != nil { + t.Fatalf("IdentitiesAddedSince() error = %v, want nil", err) + } + if index.channel != "email" { + t.Errorf("the store was asked for the channel %q, want email", index.channel) + } + if index.after != after || index.limit != 5 { + t.Errorf("the store was asked for at most %d after %+v, want at most 5 after %+v", index.limit, index.after, after) + } + if diff := cmp.Diff(index.added, got); diff != "" { + t.Errorf("IdentitiesAddedSince() mismatch (-want +got):\n%s", diff) + } + if next != index.next { + t.Errorf("next cursor = %+v, want %+v", next, index.next) + } +} + +func TestIdentitiesAddedSinceAsksForNoMoreThanItsBound(t *testing.T) { + t.Parallel() + + index := &recordingIndex{} + + _, _, err := contact.NewFeed(index, 2).IdentitiesAddedSince(t.Context(), "email", contact.Cursor{}, 5) + + if err != nil { + t.Fatalf("IdentitiesAddedSince() error = %v, want nil", err) + } + if index.limit != 2 { + t.Errorf("the store was asked for at most %d identities, want the bound of 2", index.limit) + } +} + +func TestIdentitiesAddedSinceValidatesItsInput(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + channel contact.Channel + limit int + want error + }{ + "blank channel": {channel: " ", limit: 1, want: contact.ErrEmptyChannel}, + "zero limit": {channel: "email", limit: 0, want: contact.ErrInvalidLimit}, + "below zero": {channel: "email", limit: -1, want: contact.ErrInvalidLimit}, + } + + for name, tt := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + index := &recordingIndex{} + + _, _, err := contact.NewFeed(index, 2).IdentitiesAddedSince(t.Context(), tt.channel, contact.Cursor{}, tt.limit) + + if !errors.Is(err, tt.want) { + t.Errorf("IdentitiesAddedSince() error = %v, want %v", err, tt.want) + } + if index.reads != 0 { + t.Errorf("the store was read %d times, want none", index.reads) + } + }) + } +} + +func TestIdentitiesAddedSinceReportsAStoreFailure(t *testing.T) { + t.Parallel() + + index := &recordingIndex{err: errIndexStore} + + _, _, err := contact.NewFeed(index, 2).IdentitiesAddedSince(t.Context(), "email", contact.Cursor{}, 1) + + if !errors.Is(err, errIndexStore) { + t.Errorf("IdentitiesAddedSince() error = %v, want the store failure", err) + } +} + +// serverOfTheCursor identifies the database server a unit test cursor names. +const serverOfTheCursor = 184467235 + +// waitingPage returns one identity waiting past a cursor. +func waitingPage() []contact.AddedIdentity { + return []contact.AddedIdentity{{ContactID: uuid.Must(uuid.NewV7()), Identifier: "maria@example.com"}} +} + +func TestIdentitiesAddedSinceRefusesAStaleCursorBeforeReading(t *testing.T) { + t.Parallel() + + after := contact.Cursor{Server: serverOfTheCursor, Transaction: 7, Identity: uuid.Must(uuid.NewV7())} + tests := map[string]contact.CursorCheck{ + "a cursor past the server": {Server: serverOfTheCursor, Ahead: true}, + "a cursor another server numbered": {Server: serverOfTheCursor + 1}, + } + + for name, check := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + index := &recordingIndex{added: waitingPage(), check: check} + + got, next, err := contact.NewFeed(index, 2).IdentitiesAddedSince(t.Context(), "email", after, 1) + + if !errors.Is(err, contact.ErrCursorStale) { + t.Errorf("IdentitiesAddedSince() error = %v, want %v", err, contact.ErrCursorStale) + } + if got != nil || next != after { + t.Errorf("IdentitiesAddedSince() = %+v at %+v on a refusal, want nothing at %+v", got, next, after) + } + if index.reads != 0 { + t.Errorf("the store was read %d times, want none past a stale cursor", index.reads) + } + }) + } +} + +func TestIdentitiesAddedSinceTakesTheZeroCursorOnAnyServer(t *testing.T) { + t.Parallel() + + index := &recordingIndex{added: waitingPage(), check: contact.CursorCheck{Server: serverOfTheCursor}} + + got, _, err := contact.NewFeed(index, 2).IdentitiesAddedSince(t.Context(), "email", contact.Cursor{}, 1) + + if err != nil { + t.Fatalf("IdentitiesAddedSince() error = %v, want the page", err) + } + if len(got) != 1 || index.checks != 1 { + t.Errorf("IdentitiesAddedSince() = %+v after %d cursor checks, want the page after one check", got, index.checks) + } +} + +func TestIdentitiesAddedSinceNamesTheServerInTheCursorAfterAPage(t *testing.T) { + t.Parallel() + + index := &recordingIndex{ + added: waitingPage(), + next: contact.Cursor{Transaction: 9, Identity: uuid.Must(uuid.NewV7())}, + check: contact.CursorCheck{Server: serverOfTheCursor}, + } + + _, next, err := contact.NewFeed(index, 2).IdentitiesAddedSince(t.Context(), "email", contact.Cursor{}, 1) + + if err != nil { + t.Fatalf("IdentitiesAddedSince() error = %v, want the page", err) + } + want := contact.Cursor{Server: serverOfTheCursor, Transaction: 9, Identity: index.next.Identity} + if next != want { + t.Errorf("next cursor = %+v, want %+v", next, want) + } +} + +func TestIdentitiesAddedSinceAnswersTheGivenCursorOnAnEmptyPage(t *testing.T) { + t.Parallel() + + after := contact.Cursor{Server: serverOfTheCursor, Transaction: 7, Identity: uuid.Must(uuid.NewV7())} + index := &recordingIndex{check: contact.CursorCheck{Server: serverOfTheCursor}} + + got, next, err := contact.NewFeed(index, 2).IdentitiesAddedSince(t.Context(), "email", after, 1) + + if err != nil { + t.Fatalf("IdentitiesAddedSince() error = %v, want nil", err) + } + if got != nil || next != after { + t.Errorf("IdentitiesAddedSince() = %+v at %+v, want nothing at %+v", got, next, after) + } +} + +func TestIdentitiesAddedSinceReportsAFailedCursorCheckBeforeReading(t *testing.T) { + t.Parallel() + + after := contact.Cursor{Server: serverOfTheCursor, Transaction: 7} + index := &recordingIndex{added: waitingPage(), checkErr: errIndexStore} + + got, next, err := contact.NewFeed(index, 2).IdentitiesAddedSince(t.Context(), "email", after, 1) + + if !errors.Is(err, errIndexStore) { + t.Errorf("IdentitiesAddedSince() error = %v, want the store failure", err) + } + if got != nil || next != after { + t.Errorf("IdentitiesAddedSince() = %+v at %+v on a failure, want nothing at %+v", got, next, after) + } + if index.reads != 0 { + t.Errorf("the store was read %d times, want none after a failed check", index.reads) + } +} + +func TestHorizonAnswersTheStoreHorizon(t *testing.T) { + t.Parallel() + + index := &recordingIndex{horizon: contact.Cursor{Transaction: 42}} + + got, err := contact.NewFeed(index, 2).Horizon(t.Context()) + + if err != nil { + t.Fatalf("Horizon() error = %v, want nil", err) + } + if got != index.horizon { + t.Errorf("Horizon() = %+v, want %+v", got, index.horizon) + } +} + +func TestHorizonReportsAStoreFailure(t *testing.T) { + t.Parallel() + + index := &recordingIndex{err: errIndexStore} + + if _, err := contact.NewFeed(index, 2).Horizon(t.Context()); !errors.Is(err, errIndexStore) { + t.Errorf("Horizon() error = %v, want the store failure", err) + } +} diff --git a/internal/contact/match.go b/internal/contact/match.go new file mode 100644 index 00000000..6083c375 --- /dev/null +++ b/internal/contact/match.go @@ -0,0 +1,53 @@ +// SPDX-License-Identifier: Elastic-2.0 + +package contact + +import ( + "context" + "errors" + "fmt" +) + +// ErrTooManyIdentifiers reports a batch holding more identifiers than a match answers at once. +var ErrTooManyIdentifiers = errors.New("contact: too many identifiers") + +// DefaultMatchMax is the most identifiers one match answers when the host names no bound. +const DefaultMatchMax = 100 + +// MatchStore finds the contacts owning many identities in one read. +type MatchStore interface { + MatchIdentities(ctx context.Context, channel Channel, identifiers []string) (map[string]Contact, error) +} + +// Matcher finds the contacts owning a batch of identities, refusing a batch above its bound. +type Matcher struct { + store MatchStore + bound int +} + +// NewMatcher returns a [Matcher] over store answering at most bound identifiers a call. +func NewMatcher(store MatchStore, bound int) *Matcher { + return &Matcher{store: store, bound: bound} +} + +// MatchIdentities returns the contact owning each identifier on channel that one owns, keyed by its canonical form. +func (m *Matcher) MatchIdentities( + ctx context.Context, channel Channel, identifiers []string, +) (map[string]Contact, error) { + if len(identifiers) > m.bound { + return nil, fmt.Errorf("%w: %d asked, at most %d", ErrTooManyIdentifiers, len(identifiers), m.bound) + } + normalizedChannel := normalizeChannel(channel) + if normalizedChannel == "" { + return nil, ErrEmptyChannel + } + canonical := make([]string, 0, len(identifiers)) + for _, identifier := range identifiers { + _, key, err := identityKey(normalizedChannel, identifier) + if err != nil { + return nil, err + } + canonical = append(canonical, key) + } + return m.store.MatchIdentities(ctx, normalizedChannel, canonical) +} diff --git a/internal/contact/match_test.go b/internal/contact/match_test.go new file mode 100644 index 00000000..2a10b6e2 --- /dev/null +++ b/internal/contact/match_test.go @@ -0,0 +1,161 @@ +// SPDX-License-Identifier: Elastic-2.0 + +package contact_test + +import ( + "context" + "errors" + "testing" + + "github.com/google/go-cmp/cmp" + "github.com/google/uuid" + + "github.com/gopherium/alphone/internal/contact" +) + +var _ contact.MatchStore = (*recordingIndex)(nil) + +var _ contact.FeedStore = (*recordingIndex)(nil) + +var errIndexStore = errors.New("the identity index is unreachable") + +// recordingIndex answers fixed results and records what each read asked for. +type recordingIndex struct { + matches map[string]contact.Contact + added []contact.AddedIdentity + next contact.Cursor + horizon contact.Cursor + err error + check contact.CursorCheck + checkErr error + reads int + checks int + channel contact.Channel + identifiers []string + after contact.Cursor + limit int +} + +// MatchIdentities records the channel and identifiers and answers the fixed matches. +func (r *recordingIndex) MatchIdentities( + _ context.Context, channel contact.Channel, identifiers []string, +) (map[string]contact.Contact, error) { + r.reads++ + r.channel, r.identifiers = channel, identifiers + return r.matches, r.err +} + +// IdentitiesAddedSince records the channel, cursor and limit and answers the fixed page. +func (r *recordingIndex) IdentitiesAddedSince( + _ context.Context, channel contact.Channel, after contact.Cursor, limit int, +) ([]contact.AddedIdentity, contact.Cursor, error) { + r.reads++ + r.channel, r.after, r.limit = channel, after, limit + return r.added, r.next, r.err +} + +// IdentityHorizon answers the fixed horizon. +func (r *recordingIndex) IdentityHorizon(context.Context) (contact.Cursor, error) { + r.reads++ + return r.horizon, r.err +} + +// IdentityCursorCheck counts the check and answers the fixed check. +func (r *recordingIndex) IdentityCursorCheck(context.Context, contact.Cursor) (contact.CursorCheck, error) { + r.checks++ + return r.check, r.checkErr +} + +func TestAMatchAnswersAHundredIdentifiersByDefault(t *testing.T) { + t.Parallel() + + if contact.DefaultMatchMax != 100 { + t.Errorf("DefaultMatchMax = %d, want 100", contact.DefaultMatchMax) + } +} + +func TestMatchIdentitiesAsksForTheCanonicalForms(t *testing.T) { + t.Parallel() + + maria := contact.Contact{ID: uuid.Must(uuid.NewV7()), Name: "Maria Perez"} + index := &recordingIndex{matches: map[string]contact.Contact{"maria@example.com": maria}} + + got, err := contact.NewMatcher(index, 2).MatchIdentities( + t.Context(), " Email ", []string{"Maria@Example.COM", " ADA@example.com "}, + ) + + if err != nil { + t.Fatalf("MatchIdentities() error = %v, want nil", err) + } + if index.channel != "email" { + t.Errorf("the store was asked for channel %q, want email", index.channel) + } + if diff := cmp.Diff([]string{"maria@example.com", "ada@example.com"}, index.identifiers); diff != "" { + t.Errorf("the store was asked for identifiers (-want +got):\n%s", diff) + } + if diff := cmp.Diff(index.matches, got); diff != "" { + t.Errorf("MatchIdentities() mismatch (-want +got):\n%s", diff) + } +} + +func TestMatchIdentitiesRefusesABatchAboveItsBoundWithoutReading(t *testing.T) { + t.Parallel() + + index := &recordingIndex{} + + _, err := contact.NewMatcher(index, 2).MatchIdentities( + t.Context(), "email", []string{"maria@example.com", "ada@example.com", "grace@example.com"}, + ) + + if !errors.Is(err, contact.ErrTooManyIdentifiers) { + t.Errorf("MatchIdentities() error = %v, want %v", err, contact.ErrTooManyIdentifiers) + } + if index.reads != 0 { + t.Errorf("the store was read %d times, want none", index.reads) + } +} + +func TestMatchIdentitiesValidatesItsInput(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + channel contact.Channel + identifiers []string + want error + }{ + "blank channel": { + channel: " ", identifiers: []string{"maria@example.com"}, want: contact.ErrEmptyChannel, + }, + "blank identifier": { + channel: "email", identifiers: []string{"maria@example.com", " "}, want: contact.ErrEmptyIdentifier, + }, + } + + for name, tt := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + index := &recordingIndex{} + + _, err := contact.NewMatcher(index, 2).MatchIdentities(t.Context(), tt.channel, tt.identifiers) + + if !errors.Is(err, tt.want) { + t.Errorf("MatchIdentities() error = %v, want %v", err, tt.want) + } + if index.reads != 0 { + t.Errorf("the store was read %d times, want none", index.reads) + } + }) + } +} + +func TestMatchIdentitiesReportsAStoreFailure(t *testing.T) { + t.Parallel() + + index := &recordingIndex{err: errIndexStore} + + _, err := contact.NewMatcher(index, 2).MatchIdentities(t.Context(), "email", []string{"maria@example.com"}) + + if !errors.Is(err, errIndexStore) { + t.Errorf("MatchIdentities() error = %v, want the store failure", err) + } +} diff --git a/internal/postgres/contacts.go b/internal/postgres/contacts.go index 516b7653..8290b9ee 100644 --- a/internal/postgres/contacts.go +++ b/internal/postgres/contacts.go @@ -19,6 +19,10 @@ import ( var _ contact.Store = (*ContactStore)(nil) +var _ contact.MatchStore = (*ContactStore)(nil) + +var _ contact.FeedStore = (*ContactStore)(nil) + // likeEscape is the escape character [dbkit.EscapeLike] writes before each LIKE pattern character. const likeEscape = `\` diff --git a/internal/postgres/db/models.go b/internal/postgres/db/models.go index 16b5e9e2..fe562160 100644 --- a/internal/postgres/db/models.go +++ b/internal/postgres/db/models.go @@ -45,6 +45,8 @@ type CoreContactIdentity struct { DisplayName string CreatedAt time.Time TenantID uuid.UUID + AddedXid uint64 + AddedServer int64 } type CoreTask struct { diff --git a/internal/postgres/db/queries.sql.go b/internal/postgres/db/queries.sql.go index 85cf2a34..211e8f9e 100644 --- a/internal/postgres/db/queries.sql.go +++ b/internal/postgres/db/queries.sql.go @@ -511,9 +511,19 @@ type GetIdentityParams struct { TenantID uuid.UUID } -func (q *Queries) GetIdentity(ctx context.Context, arg GetIdentityParams) (CoreContactIdentity, error) { +type GetIdentityRow struct { + ID uuid.UUID + ContactID uuid.UUID + Channel string + Identifier string + DisplayName string + CreatedAt time.Time + TenantID uuid.UUID +} + +func (q *Queries) GetIdentity(ctx context.Context, arg GetIdentityParams) (GetIdentityRow, error) { row := q.db.QueryRow(ctx, getIdentity, arg.Channel, arg.Identifier, arg.TenantID) - var i CoreContactIdentity + var i GetIdentityRow err := row.Scan( &i.ID, &i.ContactID, @@ -558,6 +568,40 @@ func (q *Queries) GetTask(ctx context.Context, arg GetTaskParams) (CoreTask, err return i, err } +const identityCursorCheck = `-- name: IdentityCursorCheck :one +SELECT (pg_control_system()).system_identifier::bigint AS server, + $1::xid8 > pg_snapshot_xmax(pg_current_snapshot()) AS ahead +` + +type IdentityCursorCheckRow struct { + Server int64 + Ahead bool +} + +func (q *Queries) IdentityCursorCheck(ctx context.Context, transaction uint64) (IdentityCursorCheckRow, error) { + row := q.db.QueryRow(ctx, identityCursorCheck, transaction) + var i IdentityCursorCheckRow + err := row.Scan(&i.Server, &i.Ahead) + return i, err +} + +const identityHorizon = `-- name: IdentityHorizon :one +SELECT (pg_control_system()).system_identifier::bigint AS server, + pg_snapshot_xmin(pg_current_snapshot())::xid8 AS horizon +` + +type IdentityHorizonRow struct { + Server int64 + Horizon uint64 +} + +func (q *Queries) IdentityHorizon(ctx context.Context) (IdentityHorizonRow, error) { + row := q.db.QueryRow(ctx, identityHorizon) + var i IdentityHorizonRow + err := row.Scan(&i.Server, &i.Horizon) + return i, err +} + const listAPITokensForUser = `-- name: ListAPITokensForUser :many SELECT id, user_id, name, token_hash, created_at, last_used_at, scopes, expires_at, tenant_id FROM core.api_tokens @@ -612,15 +656,25 @@ type ListContactIdentitiesParams struct { TenantID uuid.UUID } -func (q *Queries) ListContactIdentities(ctx context.Context, arg ListContactIdentitiesParams) ([]CoreContactIdentity, error) { +type ListContactIdentitiesRow struct { + ID uuid.UUID + ContactID uuid.UUID + Channel string + Identifier string + DisplayName string + CreatedAt time.Time + TenantID uuid.UUID +} + +func (q *Queries) ListContactIdentities(ctx context.Context, arg ListContactIdentitiesParams) ([]ListContactIdentitiesRow, error) { rows, err := q.db.Query(ctx, listContactIdentities, arg.ContactIds, arg.TenantID) if err != nil { return nil, err } defer rows.Close() - var items []CoreContactIdentity + var items []ListContactIdentitiesRow for rows.Next() { - var i CoreContactIdentity + var i ListContactIdentitiesRow if err := rows.Scan( &i.ID, &i.ContactID, @@ -770,6 +824,64 @@ func (q *Queries) ListEveryAPIToken(ctx context.Context) ([]CoreApiToken, error) return items, nil } +const listIdentitiesAddedSince = `-- name: ListIdentitiesAddedSince :many +SELECT i.id, i.contact_id, i.identifier, i.created_at, i.added_xid +FROM core.contact_identities i +WHERE i.tenant_id = $1 AND i.channel = $2::text + AND (i.added_xid, i.id) > ($3::xid8, $4::uuid) + AND i.added_xid < pg_snapshot_xmin(pg_current_snapshot()) +ORDER BY i.added_xid, i.id +LIMIT $5 +` + +type ListIdentitiesAddedSinceParams struct { + TenantID uuid.UUID + Channel string + AfterXid uint64 + AfterID uuid.UUID + RowLimit int32 +} + +type ListIdentitiesAddedSinceRow struct { + ID uuid.UUID + ContactID uuid.UUID + Identifier string + CreatedAt time.Time + AddedXid uint64 +} + +func (q *Queries) ListIdentitiesAddedSince(ctx context.Context, arg ListIdentitiesAddedSinceParams) ([]ListIdentitiesAddedSinceRow, error) { + rows, err := q.db.Query(ctx, listIdentitiesAddedSince, + arg.TenantID, + arg.Channel, + arg.AfterXid, + arg.AfterID, + arg.RowLimit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListIdentitiesAddedSinceRow + for rows.Next() { + var i ListIdentitiesAddedSinceRow + if err := rows.Scan( + &i.ID, + &i.ContactID, + &i.Identifier, + &i.CreatedAt, + &i.AddedXid, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const listTasksDueBefore = `-- name: ListTasksDueBefore :many SELECT id, assignee_id, contact_id, title, status, priority, due_on, origin_source, origin_event_id, created_at, tenant_id, due_time @@ -1128,6 +1240,51 @@ func (q *Queries) ListWorkspaceWebhookSubscriptions(ctx context.Context, tenantI return items, nil } +const matchIdentities = `-- name: MatchIdentities :many +SELECT i.identifier, c.id, c.name, c.created_at +FROM core.contact_identities i +JOIN core.contacts c ON c.id = i.contact_id AND c.tenant_id = $1 +WHERE i.tenant_id = $1 AND i.channel = $2::text AND i.identifier = ANY ($3::text[]) +` + +type MatchIdentitiesParams struct { + TenantID uuid.UUID + Channel string + Identifiers []string +} + +type MatchIdentitiesRow struct { + Identifier string + ID uuid.UUID + Name string + CreatedAt time.Time +} + +func (q *Queries) MatchIdentities(ctx context.Context, arg MatchIdentitiesParams) ([]MatchIdentitiesRow, error) { + rows, err := q.db.Query(ctx, matchIdentities, arg.TenantID, arg.Channel, arg.Identifiers) + if err != nil { + return nil, err + } + defer rows.Close() + var items []MatchIdentitiesRow + for rows.Next() { + var i MatchIdentitiesRow + if err := rows.Scan( + &i.Identifier, + &i.ID, + &i.Name, + &i.CreatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const pageContacts = `-- name: PageContacts :many SELECT c.id, c.name, c.created_at, c.tenant_id FROM core.contacts c diff --git a/internal/postgres/identities.go b/internal/postgres/identities.go index bd5ed29c..87ef6ce2 100644 --- a/internal/postgres/identities.go +++ b/internal/postgres/identities.go @@ -101,6 +101,64 @@ func (s *ContactStore) AddIdentity(ctx context.Context, identity contact.Identit return nil } +// MatchIdentities returns the contact owning each identifier on channel, keyed by identifier, leaving out the unowned. +func (s *ContactStore) MatchIdentities( + ctx context.Context, channel contact.Channel, identifiers []string, +) (map[string]contact.Contact, error) { + rows, err := s.queries.MatchIdentities(ctx, db.MatchIdentitiesParams{ + TenantID: sdk.TenantOrDefault(ctx), Channel: string(channel), Identifiers: identifiers, + }) + if err != nil { + return nil, fmt.Errorf("postgres: match identities: %w", err) + } + owners := make(map[string]contact.Contact, len(rows)) + for _, row := range rows { + owners[row.Identifier] = contact.Contact{ID: row.ID, Name: row.Name, CreatedAt: row.CreatedAt} + } + return owners, nil +} + +// IdentitiesAddedSince returns at most limit identities of channel added after the cursor and the cursor after them. +func (s *ContactStore) IdentitiesAddedSince( + ctx context.Context, channel contact.Channel, after contact.Cursor, limit int, +) ([]contact.AddedIdentity, contact.Cursor, error) { + rows, err := s.queries.ListIdentitiesAddedSince(ctx, db.ListIdentitiesAddedSinceParams{ + TenantID: sdk.TenantOrDefault(ctx), + Channel: string(channel), + AfterXid: after.Transaction, + AfterID: after.Identity, + RowLimit: int32(limit), + }) + if err != nil { + return nil, after, fmt.Errorf("postgres: list identities added since: %w", err) + } + added := make([]contact.AddedIdentity, len(rows)) + next := after + for i, row := range rows { + added[i] = contact.AddedIdentity{ContactID: row.ContactID, Identifier: row.Identifier, AddedAt: row.CreatedAt} + next = contact.Cursor{Transaction: row.AddedXid, Identity: row.ID} + } + return added, next, nil +} + +// IdentityCursorCheck returns the server's identity and whether the cursor names a transaction past its count. +func (s *ContactStore) IdentityCursorCheck(ctx context.Context, cursor contact.Cursor) (contact.CursorCheck, error) { + check, err := s.queries.IdentityCursorCheck(ctx, cursor.Transaction) + if err != nil { + return contact.CursorCheck{}, fmt.Errorf("postgres: check the identity cursor: %w", err) + } + return contact.CursorCheck{Server: check.Server, Ahead: check.Ahead}, nil +} + +// IdentityHorizon returns the cursor before every identity a transaction still open may add. +func (s *ContactStore) IdentityHorizon(ctx context.Context) (contact.Cursor, error) { + horizon, err := s.queries.IdentityHorizon(ctx) + if err != nil { + return contact.Cursor{}, fmt.Errorf("postgres: identity horizon: %w", err) + } + return contact.Cursor{Server: horizon.Server, Transaction: horizon.Horizon}, nil +} + // DeleteIdentity removes the contact's identity with the given id, or // reports [contact.ErrIdentityNotFound]. func (s *ContactStore) DeleteIdentity(ctx context.Context, contactID, identityID uuid.UUID) error { diff --git a/internal/postgres/identityfeed_test.go b/internal/postgres/identityfeed_test.go new file mode 100644 index 00000000..58c51ae4 --- /dev/null +++ b/internal/postgres/identityfeed_test.go @@ -0,0 +1,486 @@ +// SPDX-License-Identifier: Elastic-2.0 + +package postgres_test + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/gopherium/alphone/internal/contact" + "github.com/gopherium/alphone/internal/postgres" + "github.com/gopherium/alphone/internal/testdb" +) + +const ( + feedSettleDeadline = 30 * time.Second + feedPause = 20 * time.Millisecond + feedPageLimit = 50 + restoredLead = 1_000_000_000 +) + +// feedReader reads one identity feed in the tenant its context serves, failing the test on any error. +type feedReader struct { + t *testing.T + ctx context.Context + feed *contact.Feed +} + +// page returns one page of the email feed after the cursor. +func (r feedReader) page(after contact.Cursor, limit int) ([]contact.AddedIdentity, contact.Cursor) { + r.t.Helper() + answered, next, err := r.feed.IdentitiesAddedSince(r.ctx, "email", after, limit) + if err != nil { + r.t.Fatalf("IdentitiesAddedSince() error = %v, want nil", err) + } + return answered, next +} + +// settled reads the email feed after the cursor until it answers at least want identities or the deadline passes. +func (r feedReader) settled(after contact.Cursor, want int) ([]contact.AddedIdentity, contact.Cursor) { + r.t.Helper() + deadline := time.Now().Add(feedSettleDeadline) + for { + answered, next := r.page(after, feedPageLimit) + if len(answered) >= want || time.Now().After(deadline) { + return answered, next + } + time.Sleep(feedPause) + } +} + +// identifiersOf returns the identifier of every answered identity, in order. +func identifiersOf(answered []contact.AddedIdentity) []string { + identifiers := make([]string, len(answered)) + for i, added := range answered { + identifiers[i] = added.Identifier + } + return identifiers +} + +// addedEmail attaches a new email identity to the contact, failing the test on any error. +func addedEmail(t *testing.T, store *postgres.ContactStore, owner uuid.UUID, address string) { + t.Helper() + if err := store.AddIdentity(t.Context(), mustIdentity(t, owner, "email", address)); err != nil { + t.Fatalf("AddIdentity(%s) error = %v, want nil", address, err) + } +} + +func TestTheFeedAnswersIdentitiesFromEveryWritePathInTheOrderTheyWereAdded(t *testing.T) { + t.Parallel() + + store := postgres.NewContactStore(newTestPool(t)) + maria := mustContact(t, "Maria Perez") + created := mustIdentity(t, maria.ID, "email", "maria@example.com") + err := store.CreateContactWithIdentities(t.Context(), maria, []contact.Identity{ + created, mustIdentity(t, maria.ID, "phone", "+184467235"), + }) + if err != nil { + t.Fatalf("CreateContactWithIdentities() error = %v, want nil", err) + } + attached := mustIdentity(t, maria.ID, "email", "maria.perez@example.com") + if err := store.AddIdentity(t.Context(), attached); err != nil { + t.Fatalf("AddIdentity() error = %v, want nil", err) + } + if _, _, err := contact.NewResolver(store).CreateWithIdentities(t.Context(), "Ada Lovelace", + []contact.Address{{Channel: "email", Identifier: "Ada@Example.com"}}); err != nil { + t.Fatalf("CreateWithIdentities() error = %v, want nil", err) + } + imported, err := store.LookupIdentity(t.Context(), "email", "ada@example.com") + if err != nil { + t.Fatalf("LookupIdentity() error = %v, want the imported identity", err) + } + + answered, _ := feedReader{t, t.Context(), contact.NewFeed(store, feedPageLimit)}.settled(contact.Cursor{}, 3) + + want := []contact.AddedIdentity{ + {ContactID: maria.ID, Identifier: "maria@example.com", AddedAt: created.CreatedAt}, + {ContactID: maria.ID, Identifier: "maria.perez@example.com", AddedAt: attached.CreatedAt}, + {ContactID: imported.ContactID, Identifier: "ada@example.com", AddedAt: imported.CreatedAt}, + } + if diff := cmp.Diff(want, answered, cmpopts.EquateApproxTime(time.Microsecond)); diff != "" { + t.Errorf("IdentitiesAddedSince() mismatch (-want +got):\n%s", diff) + } +} + +// heldIdentity begins a transaction on its own connection that stores an email identity and stays open. +func heldIdentity(t *testing.T, pool *pgxpool.Pool, owner uuid.UUID, address string) pgx.Tx { + t.Helper() + tx, err := pool.Begin(t.Context()) + if err != nil { + t.Fatalf("beginning the held transaction: %v", err) + } + t.Cleanup(func() { _ = tx.Rollback(context.Background()) }) + held := mustIdentity(t, owner, "email", address) + if _, err := tx.Exec(t.Context(), + "INSERT INTO core.contact_identities (id, contact_id, channel, identifier, display_name, created_at)"+ + " VALUES ($1, $2, $3, $4, '', $5)", + held.ID, held.ContactID, string(held.Channel), held.Identifier, held.CreatedAt); err != nil { + t.Fatalf("storing the held identity: %v", err) + } + return tx +} + +func TestTheFeedAnswersALateCommitBeforeTheLaterIdentityItWaitedBehind(t *testing.T) { + t.Parallel() + + pool := newTestPool(t) + store := postgres.NewContactStore(pool) + reader := feedReader{t, t.Context(), contact.NewFeed(store, feedPageLimit)} + maria := mustContact(t, "Maria Perez") + if err := store.Create(t.Context(), maria); err != nil { + t.Fatalf("Create() error = %v, want nil", err) + } + held := heldIdentity(t, pool, maria.ID, "maria@example.com") + addedEmail(t, store, maria.ID, "maria.perez@example.com") + + early, cursor := reader.page(contact.Cursor{}, feedPageLimit) + if err := held.Commit(t.Context()); err != nil { + t.Fatalf("committing the held transaction: %v", err) + } + late, _ := reader.settled(cursor, 2) + + if len(early) != 0 { + t.Errorf("the feed answered %v while an older transaction was open, want it to wait", identifiersOf(early)) + } + want := []string{"maria@example.com", "maria.perez@example.com"} + if diff := cmp.Diff(want, identifiersOf(late)); diff != "" { + t.Errorf("the feed after the commit mismatch (-want +got):\n%s", diff) + } +} + +func TestTheFeedResumesFromItsCursorWithoutGapsOrRepeats(t *testing.T) { + t.Parallel() + + store := postgres.NewContactStore(newTestPool(t)) + reader := feedReader{t, t.Context(), contact.NewFeed(store, feedPageLimit)} + maria := mustContact(t, "Maria Perez") + if err := store.Create(t.Context(), maria); err != nil { + t.Fatalf("Create() error = %v, want nil", err) + } + want := []string{"a@example.com", "b@example.com", "c@example.com", "d@example.com", "e@example.com"} + for _, address := range want[:3] { + addedEmail(t, store, maria.ID, address) + } + reader.settled(contact.Cursor{}, 3) + + first, cursor := reader.page(contact.Cursor{}, 2) + second, cursor := reader.page(cursor, 2) + for _, address := range want[3:] { + addedEmail(t, store, maria.ID, address) + } + rest, cursor := reader.settled(cursor, 2) + after, unchanged := reader.page(cursor, 2) + + var got []string + for _, page := range [][]contact.AddedIdentity{first, second, rest} { + got = append(got, identifiersOf(page)...) + } + if diff := cmp.Diff(want, got); diff != "" { + t.Errorf("the pages read in turn mismatch (-want +got):\n%s", diff) + } + if len(after) != 0 || unchanged != cursor { + t.Errorf("a read past the end answered %v and moved the cursor to %+v, want nothing at %+v", + identifiersOf(after), unchanged, cursor) + } +} + +func TestAReaderStartingAtTheHorizonNeverSeesAnOlderIdentity(t *testing.T) { + t.Parallel() + + store := postgres.NewContactStore(newTestPool(t)) + feed := contact.NewFeed(store, feedPageLimit) + reader := feedReader{t, t.Context(), feed} + maria := mustContact(t, "Maria Perez") + if err := store.Create(t.Context(), maria); err != nil { + t.Fatalf("Create() error = %v, want nil", err) + } + addedEmail(t, store, maria.ID, "maria@example.com") + reader.settled(contact.Cursor{}, 1) + + horizon, err := feed.Horizon(t.Context()) + if err != nil { + t.Fatalf("Horizon() error = %v, want nil", err) + } + addedEmail(t, store, maria.ID, "maria.perez@example.com") + answered, _ := reader.settled(horizon, 1) + + if diff := cmp.Diff([]string{"maria.perez@example.com"}, identifiersOf(answered)); diff != "" { + t.Errorf("the feed from the horizon mismatch (-want +got):\n%s", diff) + } +} + +// serverTransactionEnd returns the transaction number the database server behind pool hands out next. +func serverTransactionEnd(t *testing.T, pool *pgxpool.Pool) uint64 { + t.Helper() + var end uint64 + if err := pool.QueryRow(t.Context(), "SELECT pg_snapshot_xmax(pg_current_snapshot())").Scan(&end); err != nil { + t.Fatalf("reading the next transaction number: %v", err) + } + return end +} + +// serverIdentity returns the system identifier of the database server behind pool. +func serverIdentity(t *testing.T, pool *pgxpool.Pool) int64 { + t.Helper() + var server int64 + if err := pool.QueryRow(t.Context(), "SELECT system_identifier FROM pg_control_system()").Scan(&server); err != nil { + t.Fatalf("reading the server identity: %v", err) + } + return server +} + +// serversOf returns the server that numbered each identity of the contact, keyed by identifier. +func serversOf(t *testing.T, pool *pgxpool.Pool, owner uuid.UUID) map[string]int64 { + t.Helper() + rows, err := pool.Query(t.Context(), + "SELECT identifier, added_server FROM core.contact_identities WHERE contact_id = $1", owner) + if err != nil { + t.Fatalf("reading the servers that numbered the identities: %v", err) + } + servers := map[string]int64{} + var identifier string + var server int64 + if _, err := pgx.ForEachRow(rows, []any{&identifier, &server}, func() error { + servers[identifier] = server + return nil + }); err != nil { + t.Fatalf("reading the servers that numbered the identities: %v", err) + } + return servers +} + +func TestEveryWritePathStampsAnAddressWithTheServerThatNumberedIt(t *testing.T) { + t.Parallel() + + pool := newTestPool(t) + store := postgres.NewContactStore(pool) + maria := storedOwnerOf(t, store, ownedAddress{"Maria Perez", "email", "maria@example.com"}) + addedEmail(t, store, maria.ID, "maria.perez@example.com") + + server := serverIdentity(t, pool) + want := map[string]int64{"maria@example.com": server, "maria.perez@example.com": server} + if diff := cmp.Diff(want, serversOf(t, pool, maria.ID)); diff != "" { + t.Errorf("the server of each stored address mismatch (-want +got):\n%s", diff) + } +} + +func TestTheFeedRefusesACursorPastEveryTransactionTheServerBegan(t *testing.T) { + t.Parallel() + + pool := newTestPool(t) + cursor := contact.Cursor{ + Server: serverIdentity(t, pool), + Transaction: serverTransactionEnd(t, pool) + restoredLead, + Identity: uuid.Must(uuid.NewV7()), + } + + answered, next, err := contact.NewFeed(postgres.NewContactStore(pool), feedPageLimit). + IdentitiesAddedSince(t.Context(), "email", cursor, feedPageLimit) + + if !errors.Is(err, contact.ErrCursorStale) { + t.Errorf("IdentitiesAddedSince() error = %v, want %v", err, contact.ErrCursorStale) + } + if answered != nil || next != cursor { + t.Errorf("IdentitiesAddedSince() = %v at %+v, want nothing at %+v", identifiersOf(answered), next, cursor) + } +} + +func TestTheFeedTakesACursorAtTheNextTransactionTheServerHandsOut(t *testing.T) { + t.Parallel() + + pool := newTestPool(t) + cursor := contact.Cursor{Server: serverIdentity(t, pool), Transaction: serverTransactionEnd(t, pool)} + + _, next, err := contact.NewFeed(postgres.NewContactStore(pool), feedPageLimit). + IdentitiesAddedSince(t.Context(), "email", cursor, feedPageLimit) + + if err != nil || next != cursor { + t.Errorf("IdentitiesAddedSince() = %+v, %v, want %+v and nil, the cursor Horizon answers on an idle server", + next, err, cursor) + } +} + +func TestTheFeedRefusesACursorAnotherServerNumbered(t *testing.T) { + t.Parallel() + + tests := map[string]bool{ + "when no address waits past it": false, + "when a newer address waits past it": true, + } + + for name, newer := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + pool := newTestPool(t) + store := postgres.NewContactStore(pool) + feed := contact.NewFeed(store, feedPageLimit) + here := contact.Cursor{ + Server: serverIdentity(t, pool), + Transaction: serverTransactionEnd(t, pool) - 1, + Identity: uuid.Must(uuid.NewV7()), + } + elsewhere := here + elsewhere.Server++ + if newer { + storedOwnerOf(t, store, ownedAddress{"Maria Perez", "email", "maria@example.com"}) + if waiting, _ := (feedReader{t, t.Context(), feed}).settled(here, 1); len(waiting) != 1 { + t.Fatalf("the feed past this server's cursor answered %v, want maria@example.com", identifiersOf(waiting)) + } + } + + answered, next, err := feed.IdentitiesAddedSince(t.Context(), "email", elsewhere, feedPageLimit) + + if !errors.Is(err, contact.ErrCursorStale) { + t.Errorf("IdentitiesAddedSince() error = %v, want %v", err, contact.ErrCursorStale) + } + if answered != nil || next != elsewhere { + t.Errorf("IdentitiesAddedSince() = %v at %+v, want nothing at %+v", identifiersOf(answered), next, elsewhere) + } + }) + } +} + +func TestTheFeedNamesItsServerInTheCursorsItAnswers(t *testing.T) { + t.Parallel() + + pool := newTestPool(t) + store := postgres.NewContactStore(pool) + feed := contact.NewFeed(store, feedPageLimit) + storedOwnerOf(t, store, ownedAddress{"Maria Perez", "email", "maria@example.com"}) + + answered, next := feedReader{t, t.Context(), feed}.settled(contact.Cursor{}, 1) + horizon, err := feed.Horizon(t.Context()) + + if err != nil { + t.Fatalf("Horizon() error = %v, want nil", err) + } + server := serverIdentity(t, pool) + if len(answered) != 1 || next.Server != server { + t.Errorf("a page of %v answered a cursor of the server %d, want the server %d", + identifiersOf(answered), next.Server, server) + } + if horizon.Server != server { + t.Errorf("Horizon() answered a cursor of the server %d, want the server %d", horizon.Server, server) + } +} + +// stampedPastTheServer stamps every identity of the contact with a transaction the server behind pool has not reached. +func stampedPastTheServer(t *testing.T, pool *pgxpool.Pool, owner uuid.UUID) { + t.Helper() + if _, err := pool.Exec(t.Context(), + "UPDATE core.contact_identities SET added_xid = $1 WHERE contact_id = $2", + serverTransactionEnd(t, pool)+restoredLead, owner); err != nil { + t.Fatalf("stamping the identities past the server: %v", err) + } +} + +// migratedPool returns the address of a fresh migrated test database and a pool over it, skipping in short mode. +func migratedPool(t *testing.T) (string, *pgxpool.Pool) { + t.Helper() + if testing.Short() { + t.Skip("skipping database test in short mode") + } + databaseURL := testdb.New(t, testdb.Migrator()) + pool, err := pgxpool.New(t.Context(), databaseURL) + if err != nil { + t.Fatalf("connecting pool: %v", err) + } + t.Cleanup(pool.Close) + return databaseURL, pool +} + +func TestMigrateBringsAnIdentityRestoredFromAServerFurtherAlongBackIntoTheFeed(t *testing.T) { + t.Parallel() + + databaseURL, pool := migratedPool(t) + store := postgres.NewContactStore(pool) + feed := contact.NewFeed(store, feedPageLimit) + reader := feedReader{t, t.Context(), feed} + maria := storedOwnerOf(t, store, ownedAddress{"Maria Perez", "email", "maria@example.com"}) + stampedPastTheServer(t, pool, maria.ID) + following, err := feed.Horizon(t.Context()) + if err != nil { + t.Fatalf("Horizon() error = %v, want nil", err) + } + + if err := postgres.Migrate(t.Context(), databaseURL); err != nil { + t.Fatalf("Migrate() error = %v, want nil", err) + } + fromStart, _ := reader.settled(contact.Cursor{}, 1) + fromHorizon, _ := reader.settled(following, 1) + + want := []string{"maria@example.com"} + if diff := cmp.Diff(want, identifiersOf(fromStart)); diff != "" { + t.Errorf("the feed from the start after Migrate() mismatch (-want +got):\n%s", diff) + } + if diff := cmp.Diff(want, identifiersOf(fromHorizon)); diff != "" { + t.Errorf("the feed of a reader following before Migrate() mismatch (-want +got):\n%s", diff) + } +} + +// numberedElsewhere stamps the identity with the transaction as another server numbered it. +func numberedElsewhere(t *testing.T, pool *pgxpool.Pool, identifier string, transaction uint64) { + t.Helper() + if _, err := pool.Exec(t.Context(), + "UPDATE core.contact_identities SET added_server = $1, added_xid = $2 WHERE identifier = $3", + serverIdentity(t, pool)+1, transaction, identifier); err != nil { + t.Fatalf("stamping the identity as another server numbered it: %v", err) + } +} + +func TestMigrateBringsAnAddressAnotherServerSavedBackIntoTheFeedOfAReaderThisServerAnswered(t *testing.T) { + t.Parallel() + + databaseURL, pool := migratedPool(t) + store := postgres.NewContactStore(pool) + reader := feedReader{t, t.Context(), contact.NewFeed(store, feedPageLimit)} + maria := storedOwnerOf(t, store, ownedAddress{"Maria Perez", "email", "maria@example.com"}) + _, following := reader.settled(contact.Cursor{}, 1) + addedEmail(t, store, maria.ID, "maria.perez@example.com") + numberedElsewhere(t, pool, "maria.perez@example.com", following.Transaction-1) + before, _ := reader.page(following, feedPageLimit) + + if err := postgres.Migrate(t.Context(), databaseURL); err != nil { + t.Fatalf("Migrate() error = %v, want nil", err) + } + after, _ := reader.settled(following, 1) + + if len(before) != 0 { + t.Errorf("the feed from the cursor before Migrate() answered %v, want nothing", identifiersOf(before)) + } + if diff := cmp.Diff([]string{"maria.perez@example.com"}, identifiersOf(after)); diff != "" { + t.Errorf("the feed from the cursor after Migrate() mismatch (-want +got):\n%s", diff) + } + server := serverIdentity(t, pool) + want := map[string]int64{"maria@example.com": server, "maria.perez@example.com": server} + if diff := cmp.Diff(want, serversOf(t, pool, maria.ID)); diff != "" { + t.Errorf("the server of each address after Migrate() mismatch (-want +got):\n%s", diff) + } +} + +func TestTheFeedReportsAConnectionFailure(t *testing.T) { + t.Parallel() + + pool := newTestPool(t) + store := postgres.NewContactStore(pool) + pool.Close() + + if _, _, err := store.IdentitiesAddedSince(t.Context(), "email", contact.Cursor{}, 1); err == nil { + t.Error("IdentitiesAddedSince() on a closed pool error = nil, want the failure") + } + if _, err := store.IdentityHorizon(t.Context()); err == nil { + t.Error("IdentityHorizon() on a closed pool error = nil, want the failure") + } + if _, err := store.IdentityCursorCheck(t.Context(), contact.Cursor{}); err == nil { + t.Error("IdentityCursorCheck() on a closed pool error = nil, want the failure") + } +} diff --git a/internal/postgres/identitymatch_test.go b/internal/postgres/identitymatch_test.go new file mode 100644 index 00000000..a4b69d5a --- /dev/null +++ b/internal/postgres/identitymatch_test.go @@ -0,0 +1,123 @@ +// SPDX-License-Identifier: Elastic-2.0 + +package postgres_test + +import ( + "errors" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" + + "github.com/gopherium/alphone/internal/contact" + "github.com/gopherium/alphone/internal/postgres" +) + +// matchBound is the most identifiers a matcher under test answers in one call. +const matchBound = 3 + +// ownedAddress is a contact's name beside the one address it owns. +type ownedAddress struct { + name string + channel contact.Channel + identifier string +} + +// storedOwnerOf stores a contact owning the address. +func storedOwnerOf(t *testing.T, store *postgres.ContactStore, owned ownedAddress) contact.Contact { + t.Helper() + owner := mustContact(t, owned.name) + held := mustIdentity(t, owner.ID, owned.channel, owned.identifier) + if err := store.CreateContactWithIdentity(t.Context(), owner, held); err != nil { + t.Fatalf("CreateContactWithIdentity() error = %v, want nil", err) + } + return owner +} + +func TestMatchingFindsTheOwnersOfAddressesWrittenInAnotherCase(t *testing.T) { + t.Parallel() + + store := postgres.NewContactStore(newTestPool(t)) + maria := storedOwnerOf(t, store, ownedAddress{"Maria Perez", "email", "maria.perez@example.com"}) + ada := storedOwnerOf(t, store, ownedAddress{"Ada Lovelace", "email", "ada@example.com"}) + + got, err := contact.NewMatcher(store, matchBound).MatchIdentities( + t.Context(), "email", []string{"Maria.Perez@Example.COM", " ADA@example.com "}, + ) + + if err != nil { + t.Fatalf("MatchIdentities() error = %v, want nil", err) + } + want := map[string]contact.Contact{"maria.perez@example.com": maria, "ada@example.com": ada} + if diff := cmp.Diff(want, got, cmpopts.EquateApproxTime(time.Microsecond)); diff != "" { + t.Errorf("MatchIdentities() mismatch (-want +got):\n%s", diff) + } +} + +func TestMatchingLeavesUnknownAddressesOut(t *testing.T) { + t.Parallel() + + store := postgres.NewContactStore(newTestPool(t)) + maria := storedOwnerOf(t, store, ownedAddress{"Maria Perez", "email", "maria@example.com"}) + storedOwnerOf(t, store, ownedAddress{"Grace Hopper", "phone", "+184467235"}) + + got, err := contact.NewMatcher(store, matchBound).MatchIdentities( + t.Context(), "email", []string{"maria@example.com", "nobody@example.com", "+184467235"}, + ) + + if err != nil { + t.Fatalf("MatchIdentities() error = %v, want nil", err) + } + if len(got) != 1 || got["maria@example.com"].ID != maria.ID { + t.Errorf("MatchIdentities() = %+v, want only the owner of maria@example.com", got) + } +} + +func TestMatchingAnswersABatchAtTheBound(t *testing.T) { + t.Parallel() + + store := postgres.NewContactStore(newTestPool(t)) + storedOwnerOf(t, store, ownedAddress{"Maria Perez", "email", "maria@example.com"}) + + got, err := contact.NewMatcher(store, matchBound).MatchIdentities( + t.Context(), "email", []string{"maria@example.com", "ada@example.com", "grace@example.com"}, + ) + + if err != nil { + t.Fatalf("MatchIdentities() at the bound error = %v, want nil", err) + } + if len(got) != 1 { + t.Errorf("MatchIdentities() answered %d owners, want 1", len(got)) + } +} + +func TestMatchingRefusesABatchAboveTheBound(t *testing.T) { + t.Parallel() + + store := postgres.NewContactStore(newTestPool(t)) + storedOwnerOf(t, store, ownedAddress{"Maria Perez", "email", "maria@example.com"}) + + got, err := contact.NewMatcher(store, matchBound).MatchIdentities(t.Context(), "email", []string{ + "maria@example.com", "ada@example.com", "grace@example.com", "alan@example.com", + }) + + if !errors.Is(err, contact.ErrTooManyIdentifiers) { + t.Fatalf("MatchIdentities() error = %v, want %v", err, contact.ErrTooManyIdentifiers) + } + if got != nil { + t.Errorf("MatchIdentities() = %+v beside the refusal, want nothing", got) + } +} + +func TestMatchingReportsAConnectionFailure(t *testing.T) { + t.Parallel() + + pool := newTestPool(t) + store := postgres.NewContactStore(pool) + pool.Close() + + if _, err := store.MatchIdentities(t.Context(), "email", []string{"maria@example.com"}); err == nil { + t.Error("MatchIdentities() on a closed pool error = nil, want the failure") + } +} diff --git a/internal/postgres/migrate.go b/internal/postgres/migrate.go index 6a2e33be..3640c573 100644 --- a/internal/postgres/migrate.go +++ b/internal/postgres/migrate.go @@ -15,6 +15,11 @@ import ( var migrationSource = mustSub(Migrations, "migrations") +// restampIdentities marks each identity another server numbered, or numbered past this server's count, added now. +const restampIdentities = `UPDATE core.contact_identities SET added_xid = DEFAULT, added_server = DEFAULT +WHERE added_server <> (SELECT system_identifier FROM pg_control_system()) + OR added_xid >= pg_snapshot_xmax(pg_current_snapshot())` + // Migrate applies the core schema migrations to the database at databaseURL. func Migrate(ctx context.Context, databaseURL string) error { return migrateDatabase(ctx, "pgx", databaseURL) @@ -30,7 +35,7 @@ func migrateDatabase(ctx context.Context, driverName, databaseURL string) error return migrate(ctx, db) } -// migrate runs all pending up migrations against db using the goose Postgres provider under goose's session lock. +// migrate runs all pending up migrations against db under goose's session lock, then restamps the identities. func migrate(ctx context.Context, db *sql.DB) error { locker := mustLocker(lock.NewPostgresSessionLocker()) provider, err := goose.NewProvider(goose.DialectPostgres, db, migrationSource, goose.WithSessionLocker(locker)) @@ -40,6 +45,9 @@ func migrate(ctx context.Context, db *sql.DB) error { if _, err := provider.Up(ctx); err != nil { return fmt.Errorf("postgres: apply migrations: %w", err) } + if _, err := db.ExecContext(ctx, restampIdentities); err != nil { + return fmt.Errorf("postgres: restamp the identities: %w", err) + } return nil } diff --git a/internal/postgres/migrate_test.go b/internal/postgres/migrate_test.go index 0a1a0630..995fbef9 100644 --- a/internal/postgres/migrate_test.go +++ b/internal/postgres/migrate_test.go @@ -6,6 +6,8 @@ import ( "context" "database/sql" "errors" + "fmt" + "strings" "sync" "testing" "time" @@ -176,6 +178,25 @@ func TestMigrateAppliesEachPendingStepOnceWhenTwoRunAtOnce(t *testing.T) { } } +func TestMigrateReportsARestampOfTheIdentitiesItCannotRun(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("skipping database test in short mode") + } + + databaseURL := testdb.New(t, testdb.Migrator()) + if _, err := openDatabase(t, databaseURL).ExecContext(t.Context(), + "ALTER TABLE core.contact_identities DROP COLUMN added_xid"); err != nil { + t.Fatalf("dropping the adding transaction column: %v", err) + } + + err := postgres.Migrate(t.Context(), databaseURL) + + if !strings.Contains(fmt.Sprint(err), "restamp the identities") { + t.Errorf("Migrate() over identities without their adding transaction error = %v, want the restamp failure", err) + } +} + func TestMigrateRejectsMalformedURL(t *testing.T) { t.Parallel() diff --git a/internal/postgres/migrations/00024_identity_added_transaction.sql b/internal/postgres/migrations/00024_identity_added_transaction.sql new file mode 100644 index 00000000..ac5675ad --- /dev/null +++ b/internal/postgres/migrations/00024_identity_added_transaction.sql @@ -0,0 +1,14 @@ +-- SPDX-License-Identifier: Elastic-2.0 + +-- +goose Up +ALTER TABLE core.contact_identities ADD COLUMN added_xid xid8 NOT NULL DEFAULT '0'; +ALTER TABLE core.contact_identities ALTER COLUMN added_xid SET DEFAULT pg_current_xact_id(); +ALTER TABLE core.contact_identities ADD COLUMN added_server bigint NOT NULL DEFAULT 0; +ALTER TABLE core.contact_identities ALTER COLUMN added_server SET DEFAULT (pg_control_system()).system_identifier; +CREATE INDEX contact_identities_tenant_channel_added_id_idx + ON core.contact_identities (tenant_id, channel, added_xid, id); + +-- +goose Down +DROP INDEX core.contact_identities_tenant_channel_added_id_idx; +ALTER TABLE core.contact_identities DROP COLUMN added_server; +ALTER TABLE core.contact_identities DROP COLUMN added_xid; diff --git a/internal/postgres/postgres_test.go b/internal/postgres/postgres_test.go index 7750f020..e0fddf7c 100644 --- a/internal/postgres/postgres_test.go +++ b/internal/postgres/postgres_test.go @@ -239,6 +239,120 @@ func TestMigrationsKeepTheDueTimeOnAWholeMinuteOfTheDay(t *testing.T) { } } +// identityFeedVersion is the migration stamping every identity with the transaction that added it. +const identityFeedVersion = 24 + +// identityFeedIndex is the index the identity feed reads in order. +const identityFeedIndex = "contact_identities_tenant_channel_added_id_idx" + +// addedColumns returns how many of the added_xid and added_server columns core.contact_identities holds. +func addedColumns(t *testing.T, db *sql.DB) int { + t.Helper() + var columns int + err := db.QueryRow( + `SELECT count(*) FROM information_schema.columns + WHERE table_schema = 'core' AND table_name = 'contact_identities' + AND column_name IN ('added_xid', 'added_server')`, + ).Scan(&columns) + if err != nil { + t.Fatalf("querying information_schema.columns: %v", err) + } + return columns +} + +// identityStamp is the transaction, as text, and the server that numbered one identity. +type identityStamp struct { + transaction string + server int64 +} + +// addedStamp returns the stamp of the identity the id names. +func addedStamp(t *testing.T, db *sql.DB, id uuid.UUID) identityStamp { + t.Helper() + var stamp identityStamp + if err := db.QueryRow( + "SELECT added_xid::text, added_server FROM core.contact_identities WHERE id = $1", id, + ).Scan(&stamp.transaction, &stamp.server); err != nil { + t.Fatalf("reading the stamp of the identity: %v", err) + } + return stamp +} + +// systemIdentifier returns the system identifier of the database server behind db. +func systemIdentifier(t *testing.T, db *sql.DB) int64 { + t.Helper() + var server int64 + if err := db.QueryRow("SELECT (pg_control_system()).system_identifier").Scan(&server); err != nil { + t.Fatalf("reading the server identity: %v", err) + } + return server +} + +func TestMigrationsIndexTheIdentitiesByTenantChannelAndAddingTransaction(t *testing.T) { + t.Parallel() + + definition := indexDefinition(t, newTestDB(t), "contact_identities", identityFeedIndex) + + if !strings.HasSuffix(definition, "USING btree (tenant_id, channel, added_xid, id)") { + t.Errorf("index definition = %q, want tenant, channel, adding transaction and id", definition) + } +} + +func TestAnIdentityStoredBeforeTheFeedMigrationTakesTheZeroTransaction(t *testing.T) { + t.Parallel() + + db := newTestDB(t) + provider := coreProvider(t, db) + if _, err := provider.DownTo(t.Context(), identityFeedVersion-1); err != nil { + t.Fatalf("rolling back the identity feed migration: %v", err) + } + maria := mustContact(t, "Maria Perez") + insertContact(t, db, maria) + earlier := mustIdentity(t, maria.ID, "email", "maria@example.com") + if err := insertIdentity(db, earlier); err != nil { + t.Fatalf("inserting the earlier identity: %v", err) + } + + if _, err := provider.Up(t.Context()); err != nil { + t.Fatalf("applying the identity feed migration: %v", err) + } + later := mustIdentity(t, maria.ID, "phone", "+184467235") + if err := insertIdentity(db, later); err != nil { + t.Fatalf("inserting the later identity: %v", err) + } + + if got := addedStamp(t, db, earlier.ID); got != (identityStamp{transaction: "0"}) { + t.Errorf("the earlier identity stamp = %+v, want the transaction 0 on the server 0", got) + } + server := systemIdentifier(t, db) + if got := addedStamp(t, db, later.ID); got.transaction == "0" || got.server != server { + t.Errorf("the later identity stamp = %+v, want the transaction that stored it on the server %d", got, server) + } +} + +func TestRollingTheIdentityFeedMigrationBackDropsTheColumnAndTheIndex(t *testing.T) { + t.Parallel() + + db := newTestDB(t) + + if _, err := coreProvider(t, db).DownTo(t.Context(), identityFeedVersion-1); err != nil { + t.Fatalf("rolling back the identity feed migration: %v", err) + } + + if columns := addedColumns(t, db); columns != 0 { + t.Errorf("added_xid and added_server columns after the rollback = %d, want 0", columns) + } + var indexes int + if err := db.QueryRow( + "SELECT count(*) FROM pg_indexes WHERE schemaname = 'core' AND indexname = $1", identityFeedIndex, + ).Scan(&indexes); err != nil { + t.Fatalf("querying pg_indexes: %v", err) + } + if indexes != 0 { + t.Errorf("%s after the rollback = %d, want 0", identityFeedIndex, indexes) + } +} + func TestMigrationsStoreContactWithIdentity(t *testing.T) { t.Parallel() diff --git a/internal/postgres/queries.sql b/internal/postgres/queries.sql index f47f638b..59299f18 100644 --- a/internal/postgres/queries.sql +++ b/internal/postgres/queries.sql @@ -93,6 +93,29 @@ WHERE channel = @channel AND identifier = @identifier AND tenant_id = @tenant_id DELETE FROM core.contact_identities WHERE id = $1 AND contact_id = $2 AND tenant_id = @tenant_id; +-- name: MatchIdentities :many +SELECT i.identifier, c.id, c.name, c.created_at +FROM core.contact_identities i +JOIN core.contacts c ON c.id = i.contact_id AND c.tenant_id = @tenant_id +WHERE i.tenant_id = @tenant_id AND i.channel = @channel::text AND i.identifier = ANY (@identifiers::text[]); + +-- name: ListIdentitiesAddedSince :many +SELECT i.id, i.contact_id, i.identifier, i.created_at, i.added_xid +FROM core.contact_identities i +WHERE i.tenant_id = @tenant_id AND i.channel = @channel::text + AND (i.added_xid, i.id) > (@after_xid::xid8, @after_id::uuid) + AND i.added_xid < pg_snapshot_xmin(pg_current_snapshot()) +ORDER BY i.added_xid, i.id +LIMIT @row_limit; + +-- name: IdentityHorizon :one +SELECT (pg_control_system()).system_identifier::bigint AS server, + pg_snapshot_xmin(pg_current_snapshot())::xid8 AS horizon; + +-- name: IdentityCursorCheck :one +SELECT (pg_control_system()).system_identifier::bigint AS server, + @transaction::xid8 > pg_snapshot_xmax(pg_current_snapshot()) AS ahead; + -- name: CreateTask :one INSERT INTO core.tasks (id, assignee_id, contact_id, title, status, priority, due_on, origin_source, origin_event_id, created_at, tenant_id, due_time) diff --git a/internal/postgres/tenantscope_test.go b/internal/postgres/tenantscope_test.go index 274ca99f..44ef43b7 100644 --- a/internal/postgres/tenantscope_test.go +++ b/internal/postgres/tenantscope_test.go @@ -111,6 +111,65 @@ func TestAnIdentityResolvesInsideItsTenantAlone(t *testing.T) { } } +// storedOwnerIn stores a contact owning the address in the tenant the context serves. +func storedOwnerIn(t *testing.T, store *postgres.ContactStore, ctx context.Context, owned ownedAddress) uuid.UUID { + t.Helper() + owner := storedContact(t, store, ctx, owned.name) + if err := store.AddIdentity(ctx, mustIdentity(t, owner, owned.channel, owned.identifier)); err != nil { + t.Fatalf("AddIdentity(%s) error = %v, want nil", owned.identifier, err) + } + return owner +} + +func TestAMatchStaysInsideItsTenant(t *testing.T) { + t.Parallel() + + pool := newTestPool(t) + store := postgres.NewContactStore(pool) + matcher := contact.NewMatcher(store, matchBound) + acme := standingIn(t, seededTenant(t, pool)) + mine := storedOwnerIn(t, store, acme, ownedAddress{"Maria Perez", "email", "maria@example.com"}) + theirs := storedOwnerIn(t, store, t.Context(), ownedAddress{"Ada Lovelace", "email", "ada@example.com"}) + shared := storedOwnerIn(t, store, t.Context(), ownedAddress{"Grace Hopper", "email", "maria@example.com"}) + + inside, err := matcher.MatchIdentities(acme, "email", []string{"maria@example.com", "ada@example.com"}) + if err != nil { + t.Fatalf("MatchIdentities() inside Acme error = %v, want nil", err) + } + elsewhere, err := matcher.MatchIdentities(t.Context(), "email", []string{"maria@example.com", "ada@example.com"}) + if err != nil { + t.Fatalf("MatchIdentities() elsewhere error = %v, want nil", err) + } + + if len(inside) != 1 || inside["maria@example.com"].ID != mine { + t.Errorf("MatchIdentities() inside Acme = %+v, want only its own owner of maria@example.com", inside) + } + if elsewhere["maria@example.com"].ID != shared || elsewhere["ada@example.com"].ID != theirs { + t.Errorf("MatchIdentities() elsewhere = %+v, want the other tenant's own owners", elsewhere) + } +} + +func TestTheIdentityFeedStaysInsideItsTenant(t *testing.T) { + t.Parallel() + + pool := newTestPool(t) + store := postgres.NewContactStore(pool) + feed := contact.NewFeed(store, feedPageLimit) + acme := standingIn(t, seededTenant(t, pool)) + theirs := storedOwnerIn(t, store, t.Context(), ownedAddress{"Ada Lovelace", "email", "ada@example.com"}) + mine := storedOwnerIn(t, store, acme, ownedAddress{"Maria Perez", "email", "maria@example.com"}) + + inside, _ := feedReader{t, acme, feed}.settled(contact.Cursor{}, 1) + elsewhere, _ := feedReader{t, t.Context(), feed}.settled(contact.Cursor{}, 1) + + if len(inside) != 1 || inside[0].ContactID != mine { + t.Errorf("the Acme feed answered %+v, want only its own identity", inside) + } + if len(elsewhere) != 1 || elsewhere[0].ContactID != theirs { + t.Errorf("the other tenant's feed answered %+v, want only its own identity", elsewhere) + } +} + func TestASettingStaysInsideItsTenant(t *testing.T) { t.Parallel() diff --git a/sdk/identities.go b/sdk/identities.go new file mode 100644 index 00000000..71c87cd9 --- /dev/null +++ b/sdk/identities.go @@ -0,0 +1,53 @@ +// SPDX-License-Identifier: Elastic-2.0 + +package sdk + +import ( + "context" + "errors" + "time" + + "github.com/google/uuid" +) + +// ErrTooManyIdentifiers reports a match asking for more identifiers than the host answers in one call. +var ErrTooManyIdentifiers = errors.New("sdk: too many identifiers") + +// ErrInvalidLimit reports a feed page asked to hold fewer than one identity. +var ErrInvalidLimit = errors.New("sdk: invalid limit") + +// ErrCursorStale reports a feed cursor this server cannot continue, as one saved before a restore onto another server. +var ErrCursorStale = errors.New("sdk: cursor stale") + +// ContactMatcher finds the calling tenant's contacts owning many identities in one call. +type ContactMatcher interface { + // MatchIdentities returns the contact owning each identifier one owns, keyed by the identifier's canonical form. + MatchIdentities(ctx context.Context, channel Channel, identifiers []string) (map[string]Contact, error) +} + +// IdentityCursor marks a place in the order identities were added, its zero value the place before every identity. +type IdentityCursor struct { + // Server identifies the PostgreSQL server that numbered the transaction, zero in the zero cursor. + Server int64 + // Transaction is the transaction that added the identity at the place. + Transaction uint64 + // Identity is the id of the identity at the place. + Identity uuid.UUID +} + +// AddedIdentity is one identity a contact of the calling tenant gained. +type AddedIdentity struct { + ContactID uuid.UUID + Identifier string + AddedAt time.Time +} + +// IdentityFeed lists the identities the calling tenant's contacts gain, in an order no late commit can skip. +type IdentityFeed interface { + // IdentitiesAddedSince returns at most limit identities past the cursor, none at the end, and the next cursor. + IdentitiesAddedSince( + ctx context.Context, channel Channel, cursor IdentityCursor, limit int, + ) ([]AddedIdentity, IdentityCursor, error) + // Horizon returns the cursor a new reader starts from. + Horizon(ctx context.Context) (IdentityCursor, error) +} diff --git a/sdk/sdk.go b/sdk/sdk.go index 175f79b0..dbb16a57 100644 --- a/sdk/sdk.go +++ b/sdk/sdk.go @@ -62,8 +62,14 @@ type Deps struct { Pages PageBounds Resolver ContactResolver Contacts ContactDirectory - Getenv func(string) string - Events Publisher + // Matcher finds the calling tenant's contacts owning a batch of identities. + Matcher ContactMatcher + // ContactMatchMax is the most identifiers one Matcher call takes and the most identities one feed page holds. + ContactMatchMax int + // Identities lists the identities the calling tenant's contacts gain. + Identities IdentityFeed + Getenv func(string) string + Events Publisher // Env reads the settings under the program prefix, each value trimmed. Env Env } diff --git a/sqlc.yaml b/sqlc.yaml index 94d522d2..03389472 100644 --- a/sqlc.yaml +++ b/sqlc.yaml @@ -19,3 +19,5 @@ sql: go_type: "time.Time" - db_type: "date" go_type: "time.Time" + - db_type: "xid8" + go_type: "uint64" diff --git a/test/features/features/identity-feed.feature b/test/features/features/identity-feed.feature new file mode 100644 index 00000000..204bbf19 --- /dev/null +++ b/test/features/features/identity-feed.feature @@ -0,0 +1,15 @@ +Feature: A plugin follows the addresses its workspace gains + A plugin reads the addresses added to contacts after a cursor it keeps. + Each workspace reads only the addresses of its own contacts. + + Background: + Given a running AlphOne holding a user with an API token + And the tenant "Acme" exists + And a user "acme.member@example.com" holding a token is placed in the tenant "Acme" + + Scenario: An address added to a contact appears in the feed of its own workspace only + Given the tenant "Acme" holds the contact "Maria Perez" + And a plugin reads the email feed of each tenant from its horizon + When the tenant "Acme" adds the email address "Maria.Perez@Example.com" to that contact + Then the email feed of the tenant "Acme" answers "maria.perez@example.com" for that contact + And the email feed of the default tenant answers no address diff --git a/test/features/features_test.go b/test/features/features_test.go index 16a499ab..ec18ebb8 100644 --- a/test/features/features_test.go +++ b/test/features/features_test.go @@ -127,6 +127,14 @@ func initializeMail(t *testing.T) func(*godog.ScenarioContext) { return func(sc *godog.ScenarioContext) { registerMailSteps(sc, t) } } +// initializeIdentityFeed registers the identity feed steps beside the tenant ones. +func initializeIdentityFeed(t *testing.T) func(*godog.ScenarioContext) { + return func(sc *godog.ScenarioContext) { + registerTenantSteps(sc, t) + bindIdentityFeedSteps(sc) + } +} + // initializeContactPage registers the contact page steps. func initializeContactPage(t *testing.T) func(*godog.ScenarioContext) { return func(sc *godog.ScenarioContext) { registerContactPageSteps(sc, t) } @@ -228,3 +236,7 @@ func TestContactPage(t *testing.T) { func TestAdminSettings(t *testing.T) { runFeature(t, "features/admin-settings.feature", initializeAdminSettings(t)) } + +func TestIdentityFeed(t *testing.T) { + runFeature(t, "features/identity-feed.feature", initializeIdentityFeed(t)) +} diff --git a/test/features/steps_identity_feed_test.go b/test/features/steps_identity_feed_test.go new file mode 100644 index 00000000..92e804c9 --- /dev/null +++ b/test/features/steps_identity_feed_test.go @@ -0,0 +1,217 @@ +// SPDX-License-Identifier: Elastic-2.0 + +package features_test + +import ( + "context" + "encoding/json" + "fmt" + "slices" + "time" + + "github.com/cucumber/godog" + "github.com/google/uuid" + + "github.com/gopherium/alphone/internal/contact" + "github.com/gopherium/alphone/sdk" +) + +// createContactMutation stores one contact owning no identity through the graph. +const createContactMutation = `mutation($name: String!) { createContact(name: $name) { id } }` + +// addIdentityMutation attaches one identity to a contact through the graph. +const addIdentityMutation = `mutation($contactId: UUID!, $identity: ContactIdentityInput!) { + addContactIdentity(contactId: $contactId, identity: $identity) { id } +}` + +// defaultTenant names the tenant every unplaced caller stands in. +const defaultTenant = "default" + +// feedPageLimit is the most identities the scenario plugin reads in one call. +const feedPageLimit = 50 + +// feedSettleDeadline is how long the scenario plugin waits for an identity the feed still holds back. +const feedSettleDeadline = 30 * time.Second + +// feedPause is how long the scenario plugin waits between two reads of the feed. +const feedPause = 20 * time.Millisecond + +// createdContact is the envelope the contact creating step reads. +type createdContact struct { + Data struct { + CreateContact struct { + ID string `json:"id"` + } `json:"createContact"` + } `json:"data"` +} + +// feedReader is the plugin a scenario runs over the identity feed, keeping one cursor per tenant. +type feedReader struct { + feed sdk.IdentityFeed + cursors map[string]sdk.IdentityCursor +} + +// standingIn returns ctx serving the named tenant, the default one included. +func (w *world) standingIn(ctx context.Context, tenant string) (context.Context, error) { + if tenant == defaultTenant { + return sdk.WithTenant(ctx, sdk.DefaultTenantID), nil + } + var id uuid.UUID + if err := w.pool.QueryRow(ctx, "SELECT id FROM core.tenants WHERE name = $1", tenant).Scan(&id); err != nil { + return nil, fmt.Errorf("reading the tenant %q: %w", tenant, err) + } + return sdk.WithTenant(ctx, id), nil +} + +// createContactIn stores a contact through the graph as the user placed in the tenant and remembers it. +func (w *world) createContactIn(ctx context.Context, tenant, name string) error { + secret, err := w.secretOf(tenant) + if err != nil { + return err + } + if _, err := w.operationAs(ctx, secret, createContactMutation, map[string]any{"name": name}); err != nil { + return err + } + var answer createdContact + if err := json.Unmarshal(w.answered, &answer); err != nil { + return fmt.Errorf("decoding %s: %w", w.answered, err) + } + id, err := uuid.Parse(answer.Data.CreateContact.ID) + if err != nil { + return fmt.Errorf("the tenant %q could not store %q, answered %s", tenant, name, w.answered) + } + w.lastContact = id + return nil +} + +// addEmailIn attaches an email identity to the last contact through the graph as the user placed in the tenant. +func (w *world) addEmailIn(ctx context.Context, tenant, address string) error { + secret, err := w.secretOf(tenant) + if err != nil { + return err + } + answer, err := w.operationAs(ctx, secret, addIdentityMutation, map[string]any{ + "contactId": w.lastContact.String(), + "identity": map[string]any{"channel": "email", "identifier": address}, + }) + if err != nil { + return err + } + if len(answer.Errors) > 0 { + return fmt.Errorf("the tenant %q could not add %q, answered %s", tenant, address, w.answered) + } + return nil +} + +// startFeedReader starts the scenario plugin at the feed horizon of the default tenant and every placed one. +func (w *world) startFeedReader(ctx context.Context) error { + reader := &feedReader{ + feed: scenarioFeed{feed: contact.NewFeed(w.contacts, feedPageLimit)}, + cursors: map[string]sdk.IdentityCursor{}, + } + tenants := []string{defaultTenant} + for tenant := range w.members { + tenants = append(tenants, tenant) + } + for _, tenant := range tenants { + standing, err := w.standingIn(ctx, tenant) + if err != nil { + return err + } + horizon, err := reader.feed.Horizon(standing) + if err != nil { + return fmt.Errorf("reading the horizon of %q: %w", tenant, err) + } + reader.cursors[tenant] = horizon + } + w.reader = reader + return nil +} + +// readFeedOf returns the email identities the feed answers the tenant past its cursor. +func (w *world) readFeedOf(ctx context.Context, tenant string) ([]sdk.AddedIdentity, error) { + if w.reader == nil { + return nil, fmt.Errorf("the scenario started no feed reader") + } + standing, err := w.standingIn(ctx, tenant) + if err != nil { + return nil, err + } + answered, _, err := w.reader.feed.IdentitiesAddedSince(standing, "email", w.reader.cursors[tenant], feedPageLimit) + if err != nil { + return nil, fmt.Errorf("reading the feed of %q: %w", tenant, err) + } + return answered, nil +} + +// addedTo is an identifier the feed answered beside the contact that gained it. +type addedTo struct { + identifier string + contact uuid.UUID +} + +// addedToEach returns the identifier and the contact of every answered identity, in order. +func addedToEach(answered []sdk.AddedIdentity) []addedTo { + pairs := make([]addedTo, len(answered)) + for i, added := range answered { + pairs[i] = addedTo{added.Identifier, added.ContactID} + } + return pairs +} + +// settledFeedOf reads the feed of the tenant until it answers an identity or the deadline passes. +func (w *world) settledFeedOf(ctx context.Context, tenant string) ([]sdk.AddedIdentity, error) { + deadline := time.Now().Add(feedSettleDeadline) + for { + answered, err := w.readFeedOf(ctx, tenant) + if err != nil { + return nil, err + } + if len(answered) > 0 || time.Now().After(deadline) { + return answered, nil + } + time.Sleep(feedPause) + } +} + +// bindIdentityFeedSteps binds the identity feed steps onto an already booted world. +func bindIdentityFeedSteps(sc *godog.ScenarioContext) { + sc.Given(`^the tenant "([^"]*)" holds the contact "([^"]*)"$`, func(ctx context.Context, tenant, name string) error { + return worldFrom(ctx).createContactIn(ctx, tenant, name) + }) + + sc.Given(`^a plugin reads the email feed of each tenant from its horizon$`, func(ctx context.Context) error { + return worldFrom(ctx).startFeedReader(ctx) + }) + + sc.When(`^the tenant "([^"]*)" adds the email address "([^"]*)" to that contact$`, + func(ctx context.Context, tenant, address string) error { + return worldFrom(ctx).addEmailIn(ctx, tenant, address) + }) + + sc.Then(`^the email feed of the tenant "([^"]*)" answers "([^"]*)" for that contact$`, + func(ctx context.Context, tenant, address string) error { + w := worldFrom(ctx) + answered, err := w.settledFeedOf(ctx, tenant) + if err != nil { + return err + } + want := []addedTo{{address, w.lastContact}} + if got := addedToEach(answered); !slices.Equal(got, want) { + return fmt.Errorf("the feed of %q answered %+v, want only %q for the contact %s", + tenant, answered, address, w.lastContact) + } + return nil + }) + + sc.Then(`^the email feed of the default tenant answers no address$`, func(ctx context.Context) error { + answered, err := worldFrom(ctx).readFeedOf(ctx, defaultTenant) + if err != nil { + return err + } + if len(answered) > 0 { + return fmt.Errorf("the feed of the default tenant answered %+v, want no address", answered) + } + return nil + }) +} diff --git a/test/features/world_test.go b/test/features/world_test.go index 3eb5004c..00bc3f86 100644 --- a/test/features/world_test.go +++ b/test/features/world_test.go @@ -94,6 +94,7 @@ type world struct { entryIDs map[string]string answeredIDs []string entryField string + reader *feedReader } // newWorld boots an isolated database and a real server for one scenario. @@ -353,6 +354,32 @@ func (d scenarioDirectory) CreateWithIdentities( return sdk.Contact{ID: owner.ID, Name: owner.Name}, created, nil } +// scenarioFeed bridges the scenario's identity feed to the plugin contract. +type scenarioFeed struct { + feed *contact.Feed +} + +// IdentitiesAddedSince returns at most limit [sdk.AddedIdentity] values past the cursor and the next cursor. +func (f scenarioFeed) IdentitiesAddedSince( + ctx context.Context, channel sdk.Channel, cursor sdk.IdentityCursor, limit int, +) ([]sdk.AddedIdentity, sdk.IdentityCursor, error) { + added, next, err := f.feed.IdentitiesAddedSince(ctx, contact.Channel(channel), contact.Cursor(cursor), limit) + if err != nil { + return nil, cursor, markInvalid(err) + } + answered := make([]sdk.AddedIdentity, len(added)) + for i, identity := range added { + answered[i] = sdk.AddedIdentity(identity) + } + return answered, sdk.IdentityCursor(next), nil +} + +// Horizon returns the cursor a new reader starts from. +func (f scenarioFeed) Horizon(ctx context.Context) (sdk.IdentityCursor, error) { + horizon, err := f.feed.Horizon(ctx) + return sdk.IdentityCursor(horizon), err +} + // worldFrom returns the world the scenario carries. func worldFrom(ctx context.Context) *world { return ctx.Value(worldKey{}).(*world)