Smail/test/backend/log_store_integration_test.go

package logstore

import (
	"context"
	"database/sql"
	"database/sql/driver"
	"errors"
	"fmt"
	"os"
	"strings"
	"sync"
	"testing"
	"time"

	"git.pcux.cn/xuans/Smail/db/migration"
	"git.pcux.cn/xuans/Smail/utils/logpolicy"
	mysql "github.com/go-sql-driver/mysql"
	"github.com/lib/pq"
)

// 禁止 database/sql 在断线后重建连接，以免临时表消失后误访问同名的永久表。
type logTestConnector struct {
	driver.Connector
	mu   sync.Mutex
	used bool
}

func (c *logTestConnector) Connect(ctx context.Context) (driver.Conn, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.used {
		return nil, errors.New("temporary log test connection cannot reconnect")
	}
	c.used = true
	return c.Connector.Connect(ctx)
}

// 使用连接私有的临时表，关闭连接自动回收，不迁移或清空任何既有业务表。
func TestLogStoreServerDatabaseIntegration(t *testing.T) {
	for _, test := range []struct {
		name, driver, environment string
		dialect                   migration.Dialect
	}{
		{"mysql", "mysql", "SMAIL_TEST_MYSQL_DSN", migration.DialectMySQL},
		{"postgres", "postgres", "SMAIL_TEST_POSTGRES_DSN", migration.DialectPostgres},
	} {
		t.Run(test.name, func(t *testing.T) {
			dsn := strings.TrimSpace(os.Getenv(test.environment))
			if dsn == "" {
				t.Skip(test.environment + " is not set")
			}
			var connector driver.Connector
			var err error
			if test.dialect == migration.DialectMySQL {
				configuration, parseErr := mysql.ParseDSN(dsn)
				if parseErr != nil {
					t.Fatal("test database connection configuration is invalid")
				}
				connector, err = mysql.NewConnector(configuration)
			} else {
				connector, err = pq.NewConnector(dsn)
			}
			if err != nil {
				t.Fatal("test database connection configuration is invalid")
			}
			database := sql.OpenDB(&logTestConnector{Connector: connector})
			defer database.Close()
			database.SetMaxOpenConns(1)
			database.SetMaxIdleConns(1)
			ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
			defer cancel()
			if err := database.PingContext(ctx); err != nil {
				t.Fatal("dedicated test database is unavailable")
			}
			migrations := migration.DefaultMigrations()
			var logSchema []string
			for _, item := range migrations {
				if item.Name == "create_managed_logs" {
					logSchema = item.StatementsFor(test.dialect)
					break
				}
			}
			if len(logSchema) == 0 {
				t.Fatal("managed log schema migration is missing")
			}
			for _, statement := range logSchema {
				if strings.HasPrefix(statement, "CREATE TABLE IF NOT EXISTS ") {
					statement = strings.Replace(statement, "CREATE TABLE IF NOT EXISTS ", "CREATE TEMPORARY TABLE ", 1)
				}
				if _, err := database.ExecContext(ctx, statement); err != nil {
					t.Fatal("temporary log schema creation failed")
				}
			}
			actorRoleDefinition := "VARCHAR(20) NULL"
			if _, err := database.ExecContext(ctx, "ALTER TABLE operation_log ADD COLUMN actor_role "+actorRoleDefinition); err != nil {
				t.Fatal("temporary operation log role column creation failed")
			}
			store, err := New(database, test.dialect)
			if err != nil {
				t.Fatal(err)
			}
			now := time.Now().UTC()
			cutoff := logpolicy.Cutoff(now)
			store.now = func() time.Time { return now.Add(-logpolicy.Retention) }
			for _, kind := range []string{"operation", "mail"} {
				for index := 0; index < 1003; index++ {
					at := cutoff.Add(-time.Nanosecond)
					if index >= 1001 {
						at = cutoff.Add(time.Duration(index - 1001))
					}
					entry := Entry{ID: fmt.Sprintf("%s-%d", kind, index), CreatedAt: at, Kind: kind, ActorID: "42", Action: "兼容验证", Result: "success", Payload: []byte(`{"message":"中文日志"}`)}
					if err := store.Write(ctx, entry); err != nil {
						t.Fatal(err)
					}
					if index == 1002 {
						if err := store.Write(ctx, entry); err != nil {
							t.Fatal("idempotent replay failed")
						}
					}
				}
			}
			store.now = func() time.Time { return now }
			for _, kind := range []string{"operation", "mail"} {
				items, err := store.Query(ctx, Query{Kind: kind, ActorID: "42"})
				if err != nil || len(items) != 2 || items[0].Action != "兼容验证" {
					t.Fatalf("query or retention failed: kind=%s count=%d err=%v", kind, len(items), err)
				}
			}
			deleted, err := store.Cleanup(ctx, now)
			if err != nil || deleted != 2002 {
				t.Fatalf("batch cleanup failed: deleted=%d err=%v", deleted, err)
			}
			if deleted, err := store.Cleanup(ctx, now); err != nil || deleted != 0 {
				t.Fatalf("repeated cleanup failed: deleted=%d err=%v", deleted, err)
			}
		})
	}
}