feat: add login attempt schema and storage for rate limiting

internal/integration_tests/login_attempts_test.go

@@ -0,0 +1,131 @@
+package integration_tests
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/authorizerdev/authorizer/internal/storage/schemas"
+)
+
+// TestLoginAttempts tests AddLoginAttempt, CountFailedLoginAttemptsSince, and DeleteLoginAttemptsBefore
+func TestLoginAttempts(t *testing.T) {
+	cfg := getTestConfig()
+	ts := initTestSetup(t, cfg)
+	ctx := context.Background()
+
+	email := "ratelimit-test@example.com"
+	ipAddress := "192.0.2.1"
+	now := time.Now().Unix()
+	oneHourAgo := now - 3600
+
+	t.Run("should add a failed login attempt", func(t *testing.T) {
+		attempt := &schemas.LoginAttempt{
+			Email:       email,
+			IPAddress:   ipAddress,
+			Successful:  false,
+			AttemptedAt: now,
+		}
+		err := ts.StorageProvider.AddLoginAttempt(ctx, attempt)
+		require.NoError(t, err)
+		assert.NotEmpty(t, attempt.ID)
+		assert.Equal(t, attempt.Key, attempt.ID)
+		assert.NotZero(t, attempt.CreatedAt)
+	})
+
+	t.Run("should add a successful login attempt", func(t *testing.T) {
+		attempt := &schemas.LoginAttempt{
+			Email:       email,
+			IPAddress:   ipAddress,
+			Successful:  true,
+			AttemptedAt: now,
+		}
+		err := ts.StorageProvider.AddLoginAttempt(ctx, attempt)
+		require.NoError(t, err)
+		assert.NotEmpty(t, attempt.ID)
+	})
+
+	t.Run("should count only failed attempts since timestamp", func(t *testing.T) {
+		count, err := ts.StorageProvider.CountFailedLoginAttemptsSince(ctx, email, oneHourAgo)
+		require.NoError(t, err)
+		// We added 1 failed attempt above
+		assert.GreaterOrEqual(t, count, int64(1))
+	})
+
+	t.Run("should not count successful attempts", func(t *testing.T) {
+		// Add a second email with only successful attempts
+		successEmail := "success-only@example.com"
+		attempt := &schemas.LoginAttempt{
+			Email:       successEmail,
+			IPAddress:   ipAddress,
+			Successful:  true,
+			AttemptedAt: now,
+		}
+		err := ts.StorageProvider.AddLoginAttempt(ctx, attempt)
+		require.NoError(t, err)
+
+		count, err := ts.StorageProvider.CountFailedLoginAttemptsSince(ctx, successEmail, oneHourAgo)
+		require.NoError(t, err)
+		assert.Equal(t, int64(0), count)
+	})
+
+	t.Run("should not count attempts before the since timestamp", func(t *testing.T) {
+		oldEmail := "old-attempts@example.com"
+		twoHoursAgo := now - 7200
+		attempt := &schemas.LoginAttempt{
+			Email:       oldEmail,
+			IPAddress:   ipAddress,
+			Successful:  false,
+			AttemptedAt: twoHoursAgo,
+		}
+		err := ts.StorageProvider.AddLoginAttempt(ctx, attempt)
+		require.NoError(t, err)
+
+		// Count since one hour ago — the attempt is two hours old, should not be counted
+		count, err := ts.StorageProvider.CountFailedLoginAttemptsSince(ctx, oldEmail, oneHourAgo)
+		require.NoError(t, err)
+		assert.Equal(t, int64(0), count)
+	})
+
+	t.Run("should delete attempts before timestamp", func(t *testing.T) {
+		cleanupEmail := "cleanup@example.com"
+		twoHoursAgo := now - 7200
+
+		// Add an old failed attempt
+		oldAttempt := &schemas.LoginAttempt{
+			Email:       cleanupEmail,
+			IPAddress:   ipAddress,
+			Successful:  false,
+			AttemptedAt: twoHoursAgo,
+		}
+		err := ts.StorageProvider.AddLoginAttempt(ctx, oldAttempt)
+		require.NoError(t, err)
+
+		// Add a recent failed attempt
+		recentAttempt := &schemas.LoginAttempt{
+			Email:       cleanupEmail,
+			IPAddress:   ipAddress,
+			Successful:  false,
+			AttemptedAt: now,
+		}
+		err = ts.StorageProvider.AddLoginAttempt(ctx, recentAttempt)
+		require.NoError(t, err)
+
+		// Verify both attempts count before cleanup
+		countBefore, err := ts.StorageProvider.CountFailedLoginAttemptsSince(ctx, cleanupEmail, twoHoursAgo-1)
+		require.NoError(t, err)
+		assert.GreaterOrEqual(t, countBefore, int64(2))
+
+		// Delete attempts older than one hour ago
+		err = ts.StorageProvider.DeleteLoginAttemptsBefore(ctx, oneHourAgo)
+		require.NoError(t, err)
+
+		// Only the recent attempt should remain (not older than oneHourAgo)
+		countAfter, err := ts.StorageProvider.CountFailedLoginAttemptsSince(ctx, cleanupEmail, twoHoursAgo-1)
+		require.NoError(t, err)
+		assert.Equal(t, int64(1), countAfter)
+	})
+}

internal/storage/db/arangodb/login_attempt.go

@@ -0,0 +1,66 @@
+package arangodb
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/authorizerdev/authorizer/internal/storage/schemas"
+)
+
+// AddLoginAttempt records a login attempt in the database
+func (p *provider) AddLoginAttempt(ctx context.Context, loginAttempt *schemas.LoginAttempt) error {
+	if loginAttempt.ID == "" {
+		loginAttempt.ID = uuid.New().String()
+	}
+	loginAttempt.Key = loginAttempt.ID
+	loginAttempt.CreatedAt = time.Now().Unix()
+	if loginAttempt.AttemptedAt == 0 {
+		loginAttempt.AttemptedAt = loginAttempt.CreatedAt
+	}
+	loginAttemptCollection, _ := p.db.Collection(ctx, schemas.Collections.LoginAttempt)
+	_, err := loginAttemptCollection.CreateDocument(ctx, loginAttempt)
+	return err
+}
+
+// CountFailedLoginAttemptsSince counts failed login attempts for an email since the given Unix timestamp
+func (p *provider) CountFailedLoginAttemptsSince(ctx context.Context, email string, since int64) (int64, error) {
+	query := fmt.Sprintf(
+		"RETURN LENGTH(FOR d IN %s FILTER d.email == @email AND d.successful == false AND d.attempted_at >= @since RETURN 1)",
+		schemas.Collections.LoginAttempt,
+	)
+	bindVariables := map[string]interface{}{
+		"email": email,
+		"since": since,
+	}
+	cursor, err := p.db.Query(ctx, query, bindVariables)
+	if err != nil {
+		return 0, err
+	}
+	defer cursor.Close()
+	var count int64
+	_, err = cursor.ReadDocument(ctx, &count)
+	if err != nil {
+		return 0, err
+	}
+	return count, nil
+}
+
+// DeleteLoginAttemptsBefore removes all login attempts older than the given Unix timestamp
+func (p *provider) DeleteLoginAttemptsBefore(ctx context.Context, before int64) error {
+	query := fmt.Sprintf(
+		"FOR d IN %s FILTER d.attempted_at < @before REMOVE d IN %s",
+		schemas.Collections.LoginAttempt, schemas.Collections.LoginAttempt,
+	)
+	bindVariables := map[string]interface{}{
+		"before": before,
+	}
+	cursor, err := p.db.Query(ctx, query, bindVariables)
+	if err != nil {
+		return err
+	}
+	cursor.Close()
+	return nil
+}

internal/storage/db/arangodb/provider.go

@@ -323,6 +323,28 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) {
 		Sparse: true,
 	})
 
+	// LoginAttempt collection and indexes
+	loginAttemptCollectionExists, err := arangodb.CollectionExists(ctx, schemas.Collections.LoginAttempt)
+	if err != nil {
+		return nil, err
+	}
+	if !loginAttemptCollectionExists {
+		_, err = arangodb.CreateCollection(ctx, schemas.Collections.LoginAttempt, nil)
+		if err != nil {
+			return nil, err
+		}
+	}
+	loginAttemptCollection, err := arangodb.Collection(ctx, schemas.Collections.LoginAttempt)
+	if err != nil {
+		return nil, err
+	}
+	loginAttemptCollection.EnsureHashIndex(ctx, []string{"email"}, &arangoDriver.EnsureHashIndexOptions{
+		Sparse: true,
+	})
+	loginAttemptCollection.EnsurePersistentIndex(ctx, []string{"attempted_at"}, &arangoDriver.EnsurePersistentIndexOptions{
+		Sparse: true,
+	})
+
 	return &provider{
 		config:       cfg,
 		dependencies: deps,

internal/storage/db/cassandradb/login_attempt.go

@@ -0,0 +1,81 @@
+package cassandradb
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/gocql/gocql"
+	"github.com/google/uuid"
+
+	"github.com/authorizerdev/authorizer/internal/storage/schemas"
+)
+
+// AddLoginAttempt records a login attempt in the database
+func (p *provider) AddLoginAttempt(ctx context.Context, loginAttempt *schemas.LoginAttempt) error {
+	if loginAttempt.ID == "" {
+		loginAttempt.ID = uuid.New().String()
+	}
+	loginAttempt.Key = loginAttempt.ID
+	loginAttempt.CreatedAt = time.Now().Unix()
+	if loginAttempt.AttemptedAt == 0 {
+		loginAttempt.AttemptedAt = loginAttempt.CreatedAt
+	}
+	insertQuery := fmt.Sprintf(
+		"INSERT INTO %s (id, email, ip_address, successful, attempted_at, created_at) VALUES (?, ?, ?, ?, ?, ?)",
+		KeySpace+"."+schemas.Collections.LoginAttempt,
+	)
+	return p.db.Query(insertQuery,
+		loginAttempt.ID,
+		loginAttempt.Email,
+		loginAttempt.IPAddress,
+		loginAttempt.Successful,
+		loginAttempt.AttemptedAt,
+		loginAttempt.CreatedAt,
+	).Exec()
+}
+
+// CountFailedLoginAttemptsSince counts failed login attempts for an email since the given Unix timestamp
+func (p *provider) CountFailedLoginAttemptsSince(ctx context.Context, email string, since int64) (int64, error) {
+	countQuery := fmt.Sprintf(
+		`SELECT COUNT(*) FROM %s WHERE email = ? AND successful = false AND attempted_at >= ? ALLOW FILTERING`,
+		KeySpace+"."+schemas.Collections.LoginAttempt,
+	)
+	var count int64
+	err := p.db.Query(countQuery, email, since).Consistency(gocql.One).Scan(&count)
+	if err != nil {
+		return 0, err
+	}
+	return count, nil
+}
+
+// DeleteLoginAttemptsBefore removes all login attempts older than the given Unix timestamp
+// Cassandra does not support DELETE with a range filter directly, so we first select the IDs and delete one by one.
+func (p *provider) DeleteLoginAttemptsBefore(ctx context.Context, before int64) error {
+	selectQuery := fmt.Sprintf(
+		`SELECT id FROM %s WHERE attempted_at < ? ALLOW FILTERING`,
+		KeySpace+"."+schemas.Collections.LoginAttempt,
+	)
+	scanner := p.db.Query(selectQuery, before).Iter().Scanner()
+	var ids []string
+	for scanner.Next() {
+		var id string
+		if err := scanner.Scan(&id); err != nil {
+			return err
+		}
+		ids = append(ids, id)
+	}
+	if err := scanner.Err(); err != nil {
+		return err
+	}
+	deleteQuery := fmt.Sprintf(
+		`DELETE FROM %s WHERE id = ?`,
+		KeySpace+"."+schemas.Collections.LoginAttempt,
+	)
+	for _, id := range ids {
+		if err := p.db.Query(deleteQuery, id).Exec(); err != nil {
+			return err
+		}
+	}
+	return nil
+}

internal/storage/db/cassandradb/provider.go

@@ -345,6 +345,23 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) {
 		return nil, err
 	}
 
+	// LoginAttempt table and indexes
+	loginAttemptCollectionQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id text, email text, ip_address text, successful boolean, attempted_at bigint, created_at bigint, PRIMARY KEY (id))", KeySpace, schemas.Collections.LoginAttempt)
+	err = session.Query(loginAttemptCollectionQuery).Exec()
+	if err != nil {
+		return nil, err
+	}
+	loginAttemptEmailIndex := fmt.Sprintf("CREATE INDEX IF NOT EXISTS authorizer_login_attempt_email ON %s.%s (email)", KeySpace, schemas.Collections.LoginAttempt)
+	err = session.Query(loginAttemptEmailIndex).Exec()
+	if err != nil {
+		return nil, err
+	}
+	loginAttemptAttemptedAtIndex := fmt.Sprintf("CREATE INDEX IF NOT EXISTS authorizer_login_attempt_attempted_at ON %s.%s (attempted_at)", KeySpace, schemas.Collections.LoginAttempt)
+	err = session.Query(loginAttemptAttemptedAtIndex).Exec()
+	if err != nil {
+		return nil, err
+	}
+
 	return &provider{
 		config:       cfg,
 		dependencies: deps,