fix(storage): make the cassandradb provider work on apache cassandra

AGENTS.md

@@ -71,21 +71,22 @@ Run `make fmt` before committing; CI runs `make lint`.
 ```bash
 make test                 # Full module test run; TEST_DBS=sqlite (integration tests always SQLite)
 make test-sqlite          # Same as test — explicit SQLite-only, no Docker
-make test-all-db          # All 7 DBs via Docker (postgres, sqlite, mongodb, arangodb, scylladb, dynamodb, couchbase)
+make test-all-db          # All 8 DBs via Docker (postgres, sqlite, mongodb, arangodb, scylladb, cassandradb, dynamodb, couchbase)
 make smoke                # Release e2e smoke tests (build tag `smoke`, 5m timeout)
 
 # Single-DB targets (each spins up Docker, runs tests, tears down)
 make test-postgres
 make test-mongodb
 make test-scylladb
+make test-cassandradb
 make test-arangodb
 make test-dynamodb
 make test-couchbase
 
 # Docker helpers for test-all-db
 make test-docker-up       # Start all test DB containers + Redis
 make test-cleanup         # Remove all test containers
-make test-cleanup-postgres | test-cleanup-mongodb | test-cleanup-scylladb
+make test-cleanup-postgres | test-cleanup-mongodb | test-cleanup-scylladb | test-cleanup-cassandradb
 make test-cleanup-arangodb | test-cleanup-dynamodb | test-cleanup-couchbase
 ```
 

Makefile

@@ -152,6 +152,19 @@ test-scylladb: test-cleanup-scylladb
 	docker rm -vf authorizer_scylla_db; \
 	exit $$status
 
+# The cassandradb and scylladb providers share one implementation but take
+# different keyspace-replication CQL, and only the ScyllaDB path was ever
+# tested. The first run of this target against real Apache Cassandra failed:
+# CQL LIKE needs a SASI index there, so three queries this provider issues were
+# rejected outright. Different port from scylladb on purpose, so both can run.
+test-cassandradb: test-cleanup-cassandradb
+	docker run -d --name authorizer_cassandra_db -p 9142:9042 -e MAX_HEAP_SIZE=512M -e HEAP_NEWSIZE=128M cassandra:4.1
+	sh scripts/wait-for-test-dbs.sh cassandradb && \
+	{ go clean --testcache; TEST_DBS="cassandradb" $(GO_TEST_ALL); }; \
+	status=$$?; \
+	docker rm -vf authorizer_cassandra_db; \
+	exit $$status
+
 test-arangodb: test-cleanup-arangodb
 	docker run -d --name authorizer_arangodb -p 8529:8529 -e ARANGO_NO_AUTH=1 arangodb/arangodb:3.10.3
 	@# Wait for readiness, then ALWAYS tear the container down — including when
@@ -203,7 +216,7 @@ test-all-db: test-cleanup test-docker-up
 	@# failed to start them and produced a second, misleading failure. Same
 	@# capture-status-then-clean shape the e2e-playground target already uses.
 	go clean --testcache; \
-	TEST_DBS="couchbase,postgres,sqlite,mongodb,arangodb,scylladb,dynamodb" $(GO_TEST_ALL); \
+	TEST_DBS="couchbase,postgres,sqlite,mongodb,arangodb,scylladb,cassandradb,dynamodb" $(GO_TEST_ALL); \
 	status=$$?; \
 	$(MAKE) test-cleanup; \
 	exit $$status
@@ -214,6 +227,7 @@ test-docker-up:
 	docker run -d --name authorizer_postgres -p 5434:5432 -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=postgres postgres
 	docker run -d --name authorizer_mongodb_db -p 27017:27017 mongo:4.4.15
 	docker run -d --name authorizer_scylla_db -p 9042:9042 scylladb/scylla
+	docker run -d --name authorizer_cassandra_db -p 9142:9042 -e MAX_HEAP_SIZE=512M -e HEAP_NEWSIZE=128M cassandra:4.1
 	docker run -d --name authorizer_arangodb -p 8529:8529 -e ARANGO_NO_AUTH=1 arangodb/arangodb:3.10.3
 	docker run -d --name authorizer_dynamodb -p 8000:8000 amazon/dynamodb-local:latest
 	docker run -d --name authorizer_couchbase -p 8091-8097:8091-8097 -p 11210:11210 -p 11207:11207 -p 18091-18095:18091-18095 -p 18096:18096 -p 18097:18097 couchbase:latest
@@ -224,6 +238,7 @@ test-docker-up:
 test-cleanup:
 	-docker rm -vf authorizer_postgres
 	-docker rm -vf authorizer_scylla_db
+	-docker rm -vf authorizer_cassandra_db
 	-docker rm -vf authorizer_mongodb_db
 	-docker rm -vf authorizer_arangodb
 	-docker rm -vf authorizer_dynamodb
@@ -236,6 +251,8 @@ test-cleanup-mongodb:
 	-docker rm -vf authorizer_mongodb_db
 test-cleanup-scylladb:
 	-docker rm -vf authorizer_scylla_db
+test-cleanup-cassandradb:
+	-docker rm -vf authorizer_cassandra_db
 test-cleanup-arangodb:
 	-docker rm -vf authorizer_arangodb
 test-cleanup-dynamodb:

internal/storage/db/cassandradb/session_token.go

@@ -3,6 +3,7 @@ package cassandradb
 import (
 	"context"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/gocql/gocql"
@@ -82,18 +83,26 @@ func (p *provider) DeleteSessionTokenByUserIDAndKey(ctx context.Context, userId,
 	return deleted, nil
 }
 
-// DeleteAllSessionTokensByUserID deletes all session tokens for a user ID
+// DeleteAllSessionTokensByUserID deletes all session tokens for a user ID.
+// The userId parameter can be just the user ID (e.g. "123") or the full stored
+// format (e.g. "auth_provider:123"), matching the SQL provider.
+//
+// The match is **anchored**: the exact value, or a ":<userId>" suffix. It used
+// to be an unanchored "%<userId>%" contains-match, which let a delete for "42"
+// take out tokens belonging to "142" or "provider:427" — the same defect the
+// SQL provider already fixed, left live here because no test covered it on this
+// backend.
+//
+// Matching happens in Go rather than with CQL LIKE: Apache Cassandra only
+// accepts LIKE on a SASI index and rejects it against the plain secondary index
+// this provider creates, so these two methods failed outright there. The query
+// already carried ALLOW FILTERING, i.e. a full table scan, so the only thing
+// that changes is where the comparison runs.
 func (p *provider) DeleteAllSessionTokensByUserID(ctx context.Context, userId string) error {
-	var ids []string
-	likePattern := "%" + userId + "%"
-	query := fmt.Sprintf(`SELECT id FROM %s WHERE user_id LIKE ? ALLOW FILTERING`,
-		KeySpace+"."+schemas.Collections.SessionToken)
-	iter := p.db.Query(query, likePattern).Iter()
-	var id string
-	for iter.Scan(&id) {
-		ids = append(ids, id)
-	}
-	if err := iter.Close(); err != nil {
+	ids, err := p.sessionTokenIDsMatching(func(storedUserID string) bool {
+		return storedUserID == userId || strings.HasSuffix(storedUserID, ":"+userId)
+	})
+	if err != nil {
 		return err
 	}
 	for _, id := range ids {
@@ -106,17 +115,14 @@ func (p *provider) DeleteAllSessionTokensByUserID(ctx context.Context, userId st
 }
 
 // DeleteSessionTokensByNamespace deletes all session tokens for a namespace
+// (e.g. "auth_provider"). A legitimate prefix match: stored user_id is
+// "<namespace>:<userID>". See DeleteAllSessionTokensByUserID for why the match
+// is done in Go instead of with CQL LIKE.
 func (p *provider) DeleteSessionTokensByNamespace(ctx context.Context, namespace string) error {
-	likePattern := namespace + ":%"
-	var ids []string
-	query := fmt.Sprintf(`SELECT id FROM %s WHERE user_id LIKE ? ALLOW FILTERING`,
-		KeySpace+"."+schemas.Collections.SessionToken)
-	iter := p.db.Query(query, likePattern).Iter()
-	var id string
-	for iter.Scan(&id) {
-		ids = append(ids, id)
-	}
-	if err := iter.Close(); err != nil {
+	ids, err := p.sessionTokenIDsMatching(func(storedUserID string) bool {
+		return strings.HasPrefix(storedUserID, namespace+":")
+	})
+	if err != nil {
 		return err
 	}
 	for _, id := range ids {
@@ -383,3 +389,22 @@ func (p *provider) GetAllOAuthStates(ctx context.Context) ([]*schemas.OAuthState
 	}
 	return states, nil
 }
+
+// sessionTokenIDsMatching scans the session-token table and returns the ids
+// whose stored user_id satisfies match. One full scan, same as the ALLOW
+// FILTERING query it replaces.
+func (p *provider) sessionTokenIDsMatching(match func(storedUserID string) bool) ([]string, error) {
+	query := fmt.Sprintf(`SELECT id, user_id FROM %s`, KeySpace+"."+schemas.Collections.SessionToken)
+	iter := p.db.Query(query).Iter()
+	var ids []string
+	var id, storedUserID string
+	for iter.Scan(&id, &storedUserID) {
+		if match(storedUserID) {
+			ids = append(ids, id)
+		}
+	}
+	if err := iter.Close(); err != nil {
+		return nil, err
+	}
+	return ids, nil
+}

internal/storage/db/cassandradb/webhook.go

@@ -111,17 +111,30 @@ func (p *provider) GetWebhookByID(ctx context.Context, webhookID string) (*schem
 
 // GetWebhookByEventName to get webhook by event_name
 func (p *provider) GetWebhookByEventName(ctx context.Context, eventName string) ([]*schemas.Webhook, error) {
-	query := fmt.Sprintf(`SELECT id, event_description, event_name, endpoint, headers, enabled, created_at, updated_at FROM %s WHERE event_name LIKE ? ALLOW FILTERING`, KeySpace+"."+schemas.Collections.Webhook)
-	scanner := p.db.Query(query, eventName+"%").Iter().Scanner()
+	// Prefix match in Go, not with CQL LIKE. Apache Cassandra only accepts LIKE
+	// on a SASI index — with the plain secondary index this provider creates it
+	// rejects the query outright ("LIKE restriction is only supported on
+	// properly indexed columns"), so webhook dispatch failed on every Apache
+	// Cassandra deployment. ScyllaDB accepts the same statement, which is why
+	// it went unnoticed. Costs nothing: ALLOW FILTERING already scanned the
+	// whole table, so the only change is where the comparison happens.
+	query := fmt.Sprintf(`SELECT id, event_description, event_name, endpoint, headers, enabled, created_at, updated_at FROM %s`, KeySpace+"."+schemas.Collections.Webhook)
+	scanner := p.db.Query(query).Iter().Scanner()
 	webhooks := []*schemas.Webhook{}
 	for scanner.Next() {
 		var webhook schemas.Webhook
 		err := scanner.Scan(&webhook.ID, &webhook.EventDescription, &webhook.EventName, &webhook.EndPoint, &webhook.Headers, &webhook.Enabled, &webhook.CreatedAt, &webhook.UpdatedAt)
 		if err != nil {
 			return nil, err
 		}
+		if !strings.HasPrefix(webhook.EventName, eventName) {
+			continue
+		}
 		webhooks = append(webhooks, &webhook)
 	}
+	if err := scanner.Err(); err != nil {
+		return nil, err
+	}
 	return webhooks, nil
 }
 

internal/storage/provider_test.go

@@ -28,6 +28,7 @@ var allDBTypes = []string{
 	constants.DbTypeMongoDB,
 	constants.DbTypeArangoDB,
 	constants.DbTypeScyllaDB,
+	constants.DbTypeCassandraDB,
 	constants.DbTypeCouchbaseDB,
 	constants.DbTypeDynamoDB,
 }
@@ -70,6 +71,11 @@ func getTestDBConfig(dbType string) *config.Config {
 		cfg.DatabaseURL = "http://localhost:8529"
 	case constants.DbTypeScyllaDB:
 		cfg.DatabaseURL = "127.0.0.1:9042"
+	case constants.DbTypeCassandraDB:
+		// A separate port from ScyllaDB on purpose: this provider backs both DB
+		// types and they take different keyspace-replication CQL, so a shared
+		// port would silently test one engine twice.
+		cfg.DatabaseURL = "127.0.0.1:9142"
 	case constants.DbTypeCouchbaseDB:
 		cfg.DatabaseURL = "couchbase://127.0.0.1"
 		// Allow extra time for Couchbase container to become ready in tests (test-all-db runs Couchbase last)
@@ -230,6 +236,10 @@ func TestStorageProvider(t *testing.T) {
 				testDeleteUserCascade(t, ctx, provider)
 			})
 
+			t.Run("Anchored Session Token Delete", func(t *testing.T) {
+				testAnchoredSessionTokenDelete(t, ctx, provider)
+			})
+
 			if isSQLTestDB(dbType) {
 				t.Run("SQL CRUD Correctness Fixes", func(t *testing.T) {
 					testSQLCRUDCorrectnessFixes(t, ctx, provider)
@@ -284,36 +294,6 @@ func testSQLCRUDCorrectnessFixes(t *testing.T, ctx context.Context, provider Pro
 
 	require.NoError(t, provider.DeleteUser(ctx, userA))
 
-	// #2: DeleteAllSessionTokensByUserID must delete only the target user's
-	// tokens, not another user whose stored id contains the target as a
-	// substring. otherUser ("xu"+suffix) contains targetUser ("u"+suffix).
-	suffix := uuid.New().String()
-	targetUser := "u" + suffix
-	otherUser := "xu" + suffix
-	targetStored := "auth_provider:" + targetUser
-	otherStored := "auth_provider:" + otherUser
-
-	addToken := func(storedUserID, key string) {
-		require.NoError(t, provider.AddSessionToken(ctx, &schemas.SessionToken{
-			UserID:    storedUserID,
-			KeyName:   key,
-			Token:     "tok_" + uuid.New().String(),
-			ExpiresAt: time.Now().Add(60 * time.Second).Unix(),
-		}))
-	}
-	addToken(targetStored, "session_token_key")
-	addToken(otherStored, "session_token_key")
-
-	require.NoError(t, provider.DeleteAllSessionTokensByUserID(ctx, targetUser))
-
-	_, err = provider.GetSessionTokenByUserIDAndKey(ctx, targetStored, "session_token_key")
-	assert.Error(t, err, "target user's session token should be deleted")
-	_, err = provider.GetSessionTokenByUserIDAndKey(ctx, otherStored, "session_token_key")
-	assert.NoError(t, err, "substring-matching other user's session token must survive")
-
-	// cleanup the surviving token
-	_ = provider.DeleteAllSessionTokensByUserID(ctx, otherUser)
-
 	// #3a: UpdateUser rejects a partial struct (CreatedAt == 0, i.e. never loaded).
 	_, err = provider.UpdateUser(ctx, &schemas.User{
 		ID:    uuid.New().String(),
@@ -348,6 +328,43 @@ func testSQLCRUDCorrectnessFixes(t *testing.T, ctx context.Context, provider Pro
 	require.NoError(t, provider.DeleteUser(ctx, refetched))
 }
 
+// testAnchoredSessionTokenDelete asserts DeleteAllSessionTokensByUserID deletes
+// only the target user's tokens, never another user whose stored id merely
+// contains the target as a substring. otherUser ("xu"+suffix) contains
+// targetUser ("u"+suffix).
+//
+// It runs on every backend because that is the only way this class of bug is
+// caught: the SQL provider was anchored long ago, while Cassandra/ScyllaDB kept
+// an unanchored "%<userId>%" match that deleted the other user's token too —
+// and CI stayed green because this assertion lived inside a SQL-only subtest.
+func testAnchoredSessionTokenDelete(t *testing.T, ctx context.Context, provider Provider) {
+	suffix := uuid.New().String()
+	targetUser := "u" + suffix
+	otherUser := "xu" + suffix
+	targetStored := "auth_provider:" + targetUser
+	otherStored := "auth_provider:" + otherUser
+
+	addToken := func(storedUserID string) {
+		require.NoError(t, provider.AddSessionToken(ctx, &schemas.SessionToken{
+			UserID:    storedUserID,
+			KeyName:   "session_token_key",
+			Token:     "tok_" + uuid.New().String(),
+			ExpiresAt: time.Now().Add(60 * time.Second).Unix(),
+		}))
+	}
+	addToken(targetStored)
+	addToken(otherStored)
+
+	require.NoError(t, provider.DeleteAllSessionTokensByUserID(ctx, targetUser))
+
+	_, err := provider.GetSessionTokenByUserIDAndKey(ctx, targetStored, "session_token_key")
+	assert.Error(t, err, "target user's session token should be deleted")
+	_, err = provider.GetSessionTokenByUserIDAndKey(ctx, otherStored, "session_token_key")
+	assert.NoError(t, err, "substring-matching other user's session token must survive")
+
+	_ = provider.DeleteAllSessionTokensByUserID(ctx, otherUser)
+}
+
 // testUserSearchOperations verifies ListUsers' optional case-insensitive
 // substring filter across email/given_name/family_name/nickname, and that
 // pagination + total reflect the filtered set. Every backend must honour it,
@@ -527,7 +544,15 @@ func testUserOperations(t *testing.T, ctx context.Context, provider Provider, db
 	assert.Error(t, err)
 
 	// Test GetUserByPhoneNumber
-	user.PhoneNumber = refs.NewStringRef("+1234567891")
+	//
+	// Unique numbers per run, like the emails above. These were two fixed
+	// literals, so a run that failed before the cleanup at the bottom left
+	// them in the table and every later run against the same database failed
+	// here instead with "user with given phone number already exists" — one
+	// failure cascading into a different, misleading one.
+	phoneBefore := "+1" + strings.ReplaceAll(uuid.New().String(), "-", "")[:10]
+	phoneAfter := "+2" + strings.ReplaceAll(uuid.New().String(), "-", "")[:10]
+	user.PhoneNumber = refs.NewStringRef(phoneBefore)
 	createdUser, err = provider.AddUser(ctx, user)
 	require.NoError(t, err)
 	require.NotNil(t, createdUser)
@@ -544,10 +569,19 @@ func testUserOperations(t *testing.T, ctx context.Context, provider Provider, db
 	assert.Equal(t, *user.Password, *fetchedUser.Password)
 
 	// Test UpdateUsers
+	//
+	// Filter by this user's own email rather than listing the first page
+	// unfiltered. The unfiltered form asserted the just-created user appears in
+	// Limit:10/Offset:0, which only holds while the table is nearly empty — it
+	// passes in CI (fresh container per run) and fails on every repeated local
+	// run once a few hundred rows have accumulated, looking exactly like a
+	// write that did not take effect. Scoping the query keeps both assertions
+	// (ListUsers works; the created user carries its password) independent of
+	// how much data is already there.
 	users, _, err = provider.ListUsers(ctx, &model.Pagination{
 		Limit:  10,
 		Offset: 0,
-	}, "")
+	}, *user.Email)
 	assert.NoError(t, err)
 	assert.NotNil(t, users)
 	assert.Greater(t, len(users), 0)
@@ -564,16 +598,16 @@ func testUserOperations(t *testing.T, ctx context.Context, provider Provider, db
 	require.NotNil(t, foundInList.Password, "ListUsers must not drop the stored password")
 	assert.Equal(t, *user.Password, *foundInList.Password)
 	data := map[string]interface{}{
-		"phone_number": "+3216549870",
+		"phone_number": phoneAfter,
 	}
 	err = provider.UpdateUsers(ctx, data, []string{createdUser.ID})
 	assert.NoError(t, err)
 
 	// Test GetUserByPhoneNumber after update
-	user, err = provider.GetUserByPhoneNumber(ctx, "+3216549870")
+	user, err = provider.GetUserByPhoneNumber(ctx, phoneAfter)
 	require.NoError(t, err)
 	require.NotNil(t, user)
-	assert.Equal(t, "+3216549870", *user.PhoneNumber)
+	assert.Equal(t, phoneAfter, *user.PhoneNumber)
 
 	// Cleanup: delete the user to avoid data leakage between test runs
 	err = provider.DeleteUser(ctx, user)
@@ -709,8 +743,10 @@ func testWebhookOperations(t *testing.T, ctx context.Context, provider Provider,
 
 	// Test GetWebhookByEventName
 	fetchedByEventName, err := provider.GetWebhookByEventName(ctx, webhook.EventName)
-	assert.NoError(t, err)
-	assert.NotNil(t, fetchedByEventName)
+	require.NoError(t, err)
+	// require, not assert: indexing an empty result panics, and a panic in one
+	// backend's subtest aborts the run for every backend after it.
+	require.NotEmpty(t, fetchedByEventName, "GetWebhookByEventName must find the webhook just created")
 	assert.Equal(t, created.ID, fetchedByEventName[0].ID)
 
 	// Test UpdateWebhook mutates only the changed field and preserves the rest.

scripts/wait-for-test-dbs.sh

@@ -15,7 +15,7 @@ TIMEOUT_SECONDS="${TEST_DB_WAIT_TIMEOUT:-120}"
 
 # name:port pairs. Couchbase is absent on purpose — scripts/couchbase-test.sh
 # already provisions and waits for it.
-ALL_SERVICES="redis:6380 postgres:5434 mongodb:27017 scylladb:9042 arangodb:8529 dynamodb:8000"
+ALL_SERVICES="redis:6380 postgres:5434 mongodb:27017 scylladb:9042 cassandradb:9142 arangodb:8529 dynamodb:8000"
 
 # With no arguments, wait for everything (make test-all-db). With arguments,
 # wait only for the named services, so a single-backend target does not block on