diff --git a/cmd/server/main.go b/cmd/server/main.go index 69e6afb..ab07065 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -359,7 +359,7 @@ func setupLogger(cfg *cfgpkg.Config) { func newRepository(cfg cfgpkg.StorageConfig) (output.MessageRepository, io.Closer, error) { switch strings.ToUpper(cfg.Type) { case "SQLITE": - repo, err := sqliteadapter.New(cfg.Path) + repo, err := sqliteadapter.New(cfg.Path, cfg.TableName) if err != nil { return nil, nil, fmt.Errorf("sqlite repository: %w", err) } @@ -371,6 +371,7 @@ func newRepository(cfg cfgpkg.StorageConfig) (output.MessageRepository, io.Close MaxOpenConns: cfg.MaxOpenConns, MaxIdleConns: cfg.MaxIdleConns, ConnMaxLifetime: lifetime, + TableName: cfg.TableName, }) if err != nil { return nil, nil, fmt.Errorf("mariadb repository: %w", err) diff --git a/docs/config.docker-mariadb.yaml b/docs/config.docker-mariadb.yaml index e4034bf..16c19a6 100644 --- a/docs/config.docker-mariadb.yaml +++ b/docs/config.docker-mariadb.yaml @@ -36,6 +36,7 @@ storage: maxOpenConns: 10 maxIdleConns: 5 connMaxLifetime: "5m" + tableName: messages # 메시지 테이블 이름 (기본값: messages) queue: type: FILE diff --git a/docs/config.docker-sqlite.yaml b/docs/config.docker-sqlite.yaml index 0be6ab9..0e99009 100644 --- a/docs/config.docker-sqlite.yaml +++ b/docs/config.docker-sqlite.yaml @@ -33,6 +33,7 @@ outputs: storage: type: SQLITE path: "/app/data/relaybox.db" + tableName: messages # 메시지 테이블 이름 (기본값: messages) queue: type: FILE diff --git a/docs/config.example.yaml b/docs/config.example.yaml index 981e615..f22146b 100644 --- a/docs/config.example.yaml +++ b/docs/config.example.yaml @@ -9,6 +9,8 @@ server: log: level: INFO # DEBUG, INFO, WARN, ERROR + # 주의: DEBUG 레벨은 요청 헤더(Authorization 포함) 및 전체 바디를 로그에 기록합니다. + # 운영 환경에서는 INFO 이상을 권장합니다. format: JSON # JSON, TEXT inputs: @@ -71,6 +73,7 @@ outputs: storage: type: SQLITE path: "./data/relaybox.db" + tableName: messages # 메시지 테이블 이름 (기본값: messages) queue: type: FILE diff --git a/internal/adapter/input/http/apidocs/openapi.yaml b/internal/adapter/input/http/apidocs/openapi.yaml index c5006aa..7bdbe2c 100644 --- a/internal/adapter/input/http/apidocs/openapi.yaml +++ b/internal/adapter/input/http/apidocs/openapi.yaml @@ -4,36 +4,40 @@ info: title: relaybox API version: "2026-03-20" description: | - relaybox — generic relay hub. Receives messages from monitoring apps (Beszel, Dozzle, etc.) - and forwards them to external outputs via templates. + relaybox — 범용 릴레이 허브. 모니터링 앱(Beszel, Dozzle 등)으로부터 메시지를 수신하여 템플릿을 통해 외부 출력으로 전달합니다. servers: - url: http://localhost:8080 - description: Local development + description: 로컬 개발 security: - bearerAuth: [] tags: - name: messages - description: Message management + description: 메시지 관리 - name: inputs - description: Input management + description: 입력 관리 - name: outputs - description: Output management + description: 출력 관리 - name: health - description: Health check + description: 헬스 체크 paths: /healthz: get: - summary: Health check + summary: 헬스 체크 operationId: healthz tags: [health] security: [] + x-codeSamples: + - lang: Shell + label: curl + source: | + curl http://localhost:8080/healthz responses: "200": - description: Service is healthy + description: 서비스 정상 content: application/json: schema: @@ -43,28 +47,41 @@ paths: type: string example: ok required: [status] + example: + status: ok /inputs/{inputId}/messages: post: - summary: Receive message + summary: 메시지 수신 operationId: postMessage tags: [messages] parameters: - $ref: "#/components/parameters/inputId" + x-codeSamples: + - lang: Shell + label: curl + source: | + curl -X POST http://localhost:8080/inputs/beszel/messages \ + -H "Authorization: Bearer " \ + -H "Content-Type: application/json" \ + -d '{"title":"CPU Alert","message":"CPU usage exceeded 90% on server-01"}' requestBody: required: true content: application/json: schema: type: object - description: Input-specific JSON payload + description: 입력별 JSON 페이로드 additionalProperties: true + example: + title: "CPU Alert" + message: "CPU usage exceeded 90% on server-01" responses: "201": - description: Message created + description: 메시지 생성됨 headers: Location: - description: URL of the created message + description: 생성된 메시지의 URL schema: type: string example: /inputs/beszel/messages/01J... @@ -72,6 +89,11 @@ paths: application/json: schema: $ref: "#/components/schemas/MessageCreatedResponse" + example: + id: "01JEXAMPLE000000000000001" + inputId: beszel + status: PENDING + createdAt: "2026-03-24T12:00:00Z" "401": $ref: "#/components/responses/Unauthorized" "413": @@ -80,7 +102,7 @@ paths: $ref: "#/components/responses/InternalServerError" get: - summary: List messages for an input + summary: 입력별 메시지 목록 조회 operationId: listMessages tags: [messages] parameters: @@ -95,48 +117,88 @@ paths: schema: type: integer default: 0 + x-codeSamples: + - lang: Shell + label: curl + source: | + curl "http://localhost:8080/inputs/beszel/messages?limit=20&offset=0" \ + -H "Authorization: Bearer " responses: "200": - description: Message list + description: 메시지 목록 content: application/json: schema: type: array items: $ref: "#/components/schemas/Message" + example: + - id: "01JEXAMPLE000000000000001" + inputId: beszel + status: DELIVERED + createdAt: "2026-03-24T12:00:00Z" + retryCount: 1 + lastAttemptAt: "2026-03-24T12:00:05Z" + - id: "01JEXAMPLE000000000000002" + inputId: beszel + status: PENDING + createdAt: "2026-03-24T11:55:00Z" + retryCount: 0 + lastAttemptAt: null "401": $ref: "#/components/responses/Unauthorized" /inputs/{inputId}/messages/{messageId}: get: - summary: Get message by ID + summary: 메시지 ID로 조회 operationId: getMessage tags: [messages] parameters: - $ref: "#/components/parameters/inputId" - $ref: "#/components/parameters/messageId" + x-codeSamples: + - lang: Shell + label: curl + source: | + curl http://localhost:8080/inputs/beszel/messages/01JEXAMPLE000000000000001 \ + -H "Authorization: Bearer " responses: "200": - description: Message found + description: 메시지 조회 성공 content: application/json: schema: $ref: "#/components/schemas/Message" + example: + id: "01JEXAMPLE000000000000001" + inputId: beszel + status: DELIVERED + createdAt: "2026-03-24T12:00:00Z" + retryCount: 1 + lastAttemptAt: "2026-03-24T12:00:05Z" "401": $ref: "#/components/responses/Unauthorized" "404": $ref: "#/components/responses/NotFound" patch: - summary: Update message status + summary: 메시지 상태 변경 operationId: patchMessage tags: [messages] description: | - Allowed transition: FAILED → PENDING (manual re-queue). - All other transitions return 422. + 허용 전환: FAILED → PENDING (수동 재큐잉). + 그 외 전환은 422를 반환합니다. parameters: - $ref: "#/components/parameters/inputId" - $ref: "#/components/parameters/messageId" + x-codeSamples: + - lang: Shell + label: curl + source: | + curl -X PATCH http://localhost:8080/inputs/beszel/messages/01JEXAMPLE000000000000001 \ + -H "Authorization: Bearer " \ + -H "Content-Type: application/json" \ + -d '{"status":"PENDING"}' requestBody: required: true content: @@ -148,13 +210,22 @@ paths: type: string enum: [PENDING] required: [status] + example: + status: PENDING responses: "200": - description: Message updated + description: 메시지 업데이트됨 content: application/json: schema: $ref: "#/components/schemas/Message" + example: + id: "01JEXAMPLE000000000000001" + inputId: beszel + status: PENDING + createdAt: "2026-03-24T12:00:00Z" + retryCount: 1 + lastAttemptAt: "2026-03-24T12:00:05Z" "401": $ref: "#/components/responses/Unauthorized" "404": @@ -164,69 +235,106 @@ paths: /inputs: get: - summary: List inputs + summary: 입력 목록 조회 operationId: listInputs security: [] tags: [inputs] + x-codeSamples: + - lang: Shell + label: curl + source: | + curl http://localhost:8080/inputs responses: "200": - description: Input list + description: 입력 목록 content: application/json: schema: type: array items: $ref: "#/components/schemas/Input" + example: + - id: beszel + - id: dozzle /inputs/{inputId}: get: - summary: Get input by ID + summary: 입력 ID로 조회 operationId: getInput security: [] tags: [inputs] parameters: - $ref: "#/components/parameters/inputId" + x-codeSamples: + - lang: Shell + label: curl + source: | + curl http://localhost:8080/inputs/beszel responses: "200": - description: Input found + description: 입력 조회 성공 content: application/json: schema: $ref: "#/components/schemas/Input" + example: + id: beszel "404": $ref: "#/components/responses/NotFound" /outputs: get: - summary: List outputs + summary: 출력 목록 조회 operationId: listOutputs security: [] tags: [outputs] + x-codeSamples: + - lang: Shell + label: curl + source: | + curl http://localhost:8080/outputs responses: "200": - description: Output list + description: 출력 목록 content: application/json: schema: type: array items: $ref: "#/components/schemas/Output" + example: + - id: ops-webhook + type: WEBHOOK + url: "https://hooks.example.com/xyz" + retryCount: 3 + retryDelayMs: 1000 /outputs/{outputId}: get: - summary: Get output by ID + summary: 출력 ID로 조회 operationId: getOutput security: [] tags: [outputs] parameters: - $ref: "#/components/parameters/outputId" + x-codeSamples: + - lang: Shell + label: curl + source: | + curl http://localhost:8080/outputs/ops-webhook responses: "200": - description: Output found + description: 출력 조회 성공 content: application/json: schema: $ref: "#/components/schemas/Output" + example: + id: ops-webhook + type: WEBHOOK + url: "https://hooks.example.com/xyz" + retryCount: 3 + retryDelayMs: 1000 "404": $ref: "#/components/responses/NotFound" @@ -235,7 +343,7 @@ components: bearerAuth: type: http scheme: bearer - description: Input-specific bearer token from config + description: 입력별 Bearer 토큰 (config에서 설정) parameters: inputId: @@ -265,9 +373,9 @@ components: type: string enum: [PENDING, DELIVERED, FAILED] description: | - Message delivery status. - Transitions: PENDING → DELIVERED (success), PENDING → FAILED (retries exhausted), - FAILED → PENDING (manual re-queue via PATCH). + 메시지 전달 상태. + 전환: PENDING → DELIVERED (성공), PENDING → FAILED (재시도 소진), + FAILED → PENDING (PATCH를 통한 수동 재큐잉). Message: type: object @@ -284,7 +392,7 @@ components: format: date-time retryCount: type: integer - description: Number of delivery attempts + description: 전달 시도 횟수 lastAttemptAt: type: ["string", "null"] format: date-time @@ -329,7 +437,7 @@ components: Problem: type: object - description: RFC 7807/9457 Problem Details + description: RFC 7807/9457 Problem Details 오류 응답 properties: type: type: string @@ -349,32 +457,57 @@ components: responses: Unauthorized: - description: Missing or invalid bearer token + description: Bearer 토큰 누락 또는 유효하지 않음 content: application/problem+json: schema: $ref: "#/components/schemas/Problem" + example: + type: /errors/Unauthorized + title: Unauthorized + status: 401 + detail: "invalid or missing token for input: beszel" NotFound: - description: Resource not found + description: 리소스를 찾을 수 없음 content: application/problem+json: schema: $ref: "#/components/schemas/Problem" + example: + type: /errors/NotFound + title: Not Found + status: 404 + detail: "message not found" PayloadTooLarge: - description: Request body exceeds 1MB limit + description: 요청 본문이 1MB 제한 초과 content: application/problem+json: schema: $ref: "#/components/schemas/Problem" + example: + type: /errors/PayloadTooLarge + title: Payload Too Large + status: 413 + detail: "request body exceeds 1MB limit" UnprocessableEntity: - description: Invalid status transition + description: 유효하지 않은 상태 전환 content: application/problem+json: schema: $ref: "#/components/schemas/Problem" + example: + type: /errors/UnprocessableEntity + title: Unprocessable Entity + status: 422 + detail: "only status=\"PENDING\" is accepted" InternalServerError: - description: Unexpected server error + description: 예기치 않은 서버 오류 content: application/problem+json: schema: $ref: "#/components/schemas/Problem" + example: + type: /errors/InternalServerError + title: Internal Server Error + status: 500 + detail: "unexpected error occurred" diff --git a/internal/adapter/input/http/handler.go b/internal/adapter/input/http/handler.go index b8c4289..7384c57 100644 --- a/internal/adapter/input/http/handler.go +++ b/internal/adapter/input/http/handler.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "io" + "log/slog" "net/http" "strconv" "time" @@ -54,12 +55,25 @@ func (h *Handler) PostMessage(w http.ResponseWriter, r *http.Request) { return } + slog.Debug("request received", + "input", inputID, + "headers", r.Header, + "body", string(body), + ) + messageID, err := h.receiveUC.Receive(r.Context(), resolvedInputID, r.Header.Get("Content-Type"), body) if err != nil { mapError(w, r, err) return } + slog.Info("message received", + "input", inputID, + "messageID", messageID, + "contentType", r.Header.Get("Content-Type"), + "payloadSize", len(body), + ) + resp := map[string]any{ "id": messageID, "inputId": inputID, diff --git a/internal/adapter/input/http/handler_test.go b/internal/adapter/input/http/handler_test.go index cf4d717..1eae8b0 100644 --- a/internal/adapter/input/http/handler_test.go +++ b/internal/adapter/input/http/handler_test.go @@ -3,6 +3,7 @@ package http_test import ( "context" "encoding/json" + "log/slog" "net/http" "net/http/httptest" "strings" @@ -11,6 +12,7 @@ import ( httpadapter "relaybox/internal/adapter/input/http" inputport "relaybox/internal/application/port/input" "relaybox/internal/domain" + "relaybox/internal/testutil" ) type mockUseCase struct { @@ -735,6 +737,47 @@ func TestHandler_GetOutput_NotFound(t *testing.T) { } } +func TestPostMessage_LogsInfoOnSuccess(t *testing.T) { + h := &testutil.CaptureHandler{} + orig := slog.Default() + slog.SetDefault(slog.New(h)) + defer slog.SetDefault(orig) + + router := newTestRouter(func(_ context.Context, _ string, _ string, _ []byte) (string, error) { + return "01JTEST00000000000000000", nil + }, nil) + req := httptest.NewRequest(http.MethodPost, "/inputs/beszel/messages", strings.NewReader(`{"level":"critical"}`)) + req.Header.Set("Authorization", "Bearer test-token") + req.Header.Set("Content-Type", "application/json") + + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + if w.Code != http.StatusCreated { + t.Fatalf("status = %d, want 201", w.Code) + } + + var found bool + for _, rec := range h.Records() { + if rec.Level == slog.LevelInfo && rec.Msg == "message received" { + if _, ok := rec.Attrs["input"]; !ok { + t.Error("log record missing key: input") + } + if _, ok := rec.Attrs["messageID"]; !ok { + t.Error("log record missing key: messageID") + } + if _, ok := rec.Attrs["payloadSize"]; !ok { + t.Error("log record missing key: payloadSize") + } + found = true + break + } + } + if !found { + t.Error(`expected INFO "message received" log record not found`) + } +} + func TestDocs_HTML(t *testing.T) { router := newTestRouter(func(_ context.Context, _ string, _ string, _ []byte) (string, error) { return "", nil diff --git a/internal/adapter/output/mariadb/repository.go b/internal/adapter/output/mariadb/repository.go index 4c8ebc7..e5a3ded 100644 --- a/internal/adapter/output/mariadb/repository.go +++ b/internal/adapter/output/mariadb/repository.go @@ -8,7 +8,6 @@ import ( "time" _ "github.com/go-sql-driver/mysql" - "relaybox/internal/adapter/output/mariadb/db" output "relaybox/internal/application/port/output" "relaybox/internal/domain" ) @@ -22,16 +21,22 @@ type Config struct { MaxOpenConns int MaxIdleConns int ConnMaxLifetime time.Duration + TableName string } // Repository implements port/output.MessageRepository for MariaDB. type Repository struct { - queries *db.Queries - sqlDB *sql.DB + sqlDB *sql.DB + tableName string } // New opens a MariaDB connection, applies connection pool settings, and creates the schema. func New(cfg Config) (*Repository, error) { + tableName := cfg.TableName + if tableName == "" { + tableName = "messages" + } + dsn := ensureParseTime(cfg.DSN) sqlDB, err := sql.Open("mysql", dsn) @@ -54,25 +59,29 @@ func New(cfg Config) (*Repository, error) { sqlDB.SetConnMaxLifetime(cfg.ConnMaxLifetime) } - if _, err := sqlDB.Exec(schemaSQL); err != nil { + if _, err := sqlDB.Exec(buildSchemaSQL(tableName)); err != nil { sqlDB.Close() return nil, fmt.Errorf("apply schema: %w", err) } - return &Repository{queries: db.New(sqlDB), sqlDB: sqlDB}, nil + return &Repository{sqlDB: sqlDB, tableName: tableName}, nil } func (r *Repository) Close() error { return r.sqlDB.Close() } +func (r *Repository) buildSQL(query string) string { + return strings.ReplaceAll(query, "messages", r.tableName) +} + func (r *Repository) Save(ctx context.Context, m domain.Message) error { - err := r.queries.InsertMessage(ctx, db.InsertMessageParams{ - ID: m.ID, - Version: int32(m.Version), - Input: m.Input, - Payload: []byte(m.Payload), - CreatedAt: m.CreatedAt.UTC(), - Status: string(m.Status), - RetryCount: int32(m.RetryCount), - }) + _, err := r.sqlDB.ExecContext(ctx, r.buildSQL(sqlInsertMessage), + m.ID, + int32(m.Version), + m.Input, + []byte(m.Payload), + m.CreatedAt.UTC(), + string(m.Status), + int32(m.RetryCount), + ) if err != nil { return fmt.Errorf("save message: %w", err) } @@ -81,12 +90,12 @@ func (r *Repository) Save(ctx context.Context, m domain.Message) error { func (r *Repository) UpdateDeliveryState(ctx context.Context, id string, status domain.MessageStatus, retryCount int, lastAttemptAt time.Time) error { t := lastAttemptAt.UTC() - err := r.queries.UpdateDeliveryState(ctx, db.UpdateDeliveryStateParams{ - Status: string(status), - RetryCount: int32(retryCount), - LastAttemptAt: sql.NullTime{Time: t, Valid: true}, - ID: id, - }) + _, err := r.sqlDB.ExecContext(ctx, r.buildSQL(sqlUpdateDeliveryState), + string(status), + int32(retryCount), + sql.NullTime{Time: t, Valid: true}, + id, + ) if err != nil { return fmt.Errorf("update delivery state: %w", err) } @@ -94,25 +103,31 @@ func (r *Repository) UpdateDeliveryState(ctx context.Context, id string, status } func (r *Repository) FindByID(ctx context.Context, id string) (domain.Message, error) { - row, err := r.queries.GetMessageByID(ctx, id) + row := r.sqlDB.QueryRowContext(ctx, r.buildSQL(sqlGetMessageByID), id) + m, err := scanMessage(row) if err != nil { return domain.Message{}, fmt.Errorf("find message %q: %w", id, err) } - return toMessage(row), nil + return m, nil } func (r *Repository) FindByInput(ctx context.Context, inputID string, limit, offset int) ([]domain.Message, error) { - rows, err := r.queries.ListMessagesByInput(ctx, db.ListMessagesByInputParams{ - Input: inputID, - Limit: int32(limit), - Offset: int32(offset), - }) + rows, err := r.sqlDB.QueryContext(ctx, r.buildSQL(sqlListMessagesByInput), inputID, int32(limit), int32(offset)) if err != nil { return nil, fmt.Errorf("list messages: %w", err) } - messages := make([]domain.Message, 0, len(rows)) - for _, row := range rows { - messages = append(messages, toMessage(row)) + defer rows.Close() + + messages := make([]domain.Message, 0) + for rows.Next() { + m, err := scanMessageRow(rows) + if err != nil { + return nil, fmt.Errorf("scan message: %w", err) + } + messages = append(messages, m) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("rows error: %w", err) } return messages, nil } @@ -122,12 +137,12 @@ func (r *Repository) DeleteOlderThan(ctx context.Context, cutoff time.Time, stat var args []any if len(statuses) == 0 { - query = `DELETE FROM messages WHERE created_at < ?` + query = `DELETE FROM ` + r.tableName + ` WHERE created_at < ?` args = []any{cutoff.UTC()} } else { placeholders := strings.Repeat("?,", len(statuses)) placeholders = placeholders[:len(placeholders)-1] - query = `DELETE FROM messages WHERE created_at < ? AND status IN (` + placeholders + `)` + query = `DELETE FROM ` + r.tableName + ` WHERE created_at < ? AND status IN (` + placeholders + `)` args = make([]any, 0, 1+len(statuses)) args = append(args, cutoff.UTC()) for _, s := range statuses { @@ -146,21 +161,74 @@ func (r *Repository) DeleteOlderThan(ctx context.Context, cutoff time.Time, stat return n, nil } -func toMessage(row db.Message) domain.Message { - m := domain.Message{ - ID: row.ID, - Version: int(row.Version), - Input: row.Input, - Payload: domain.RawPayload(row.Payload), - CreatedAt: row.CreatedAt, - Status: domain.MessageStatus(row.Status), - RetryCount: int(row.RetryCount), - } - if row.LastAttemptAt.Valid { - t := row.LastAttemptAt.Time +type rowScanner interface { + Scan(dest ...any) error +} + +func scanMessage(row rowScanner) (domain.Message, error) { + var m domain.Message + var lastAttemptAt sql.NullTime + var createdAt time.Time + var status string + var version, retryCount int32 + var payload []byte + + err := row.Scan( + &m.ID, + &version, + &m.Input, + &payload, + &createdAt, + &status, + &retryCount, + &lastAttemptAt, + ) + if err != nil { + return domain.Message{}, err + } + m.Version = int(version) + m.Payload = domain.RawPayload(payload) + m.CreatedAt = createdAt + m.Status = domain.MessageStatus(status) + m.RetryCount = int(retryCount) + if lastAttemptAt.Valid { + t := lastAttemptAt.Time + m.LastAttemptAt = &t + } + return m, nil +} + +func scanMessageRow(rows *sql.Rows) (domain.Message, error) { + var m domain.Message + var lastAttemptAt sql.NullTime + var createdAt time.Time + var status string + var version, retryCount int32 + var payload []byte + + err := rows.Scan( + &m.ID, + &version, + &m.Input, + &payload, + &createdAt, + &status, + &retryCount, + &lastAttemptAt, + ) + if err != nil { + return domain.Message{}, err + } + m.Version = int(version) + m.Payload = domain.RawPayload(payload) + m.CreatedAt = createdAt + m.Status = domain.MessageStatus(status) + m.RetryCount = int(retryCount) + if lastAttemptAt.Valid { + t := lastAttemptAt.Time m.LastAttemptAt = &t } - return m + return m, nil } // ensureParseTime appends parseTime=true to the DSN if not already present. @@ -175,7 +243,24 @@ func ensureParseTime(dsn string) string { return dsn + "?parseTime=true" } -const schemaSQL = ` +const ( + sqlInsertMessage = ` +INSERT INTO messages (id, version, input, payload, created_at, status, retry_count) +VALUES (?, ?, ?, ?, ?, ?, ?)` + + sqlGetMessageByID = ` +SELECT id, version, input, payload, created_at, status, retry_count, last_attempt_at +FROM messages WHERE id=?` + + sqlListMessagesByInput = ` +SELECT id, version, input, payload, created_at, status, retry_count, last_attempt_at +FROM messages WHERE input=? ORDER BY created_at DESC LIMIT ? OFFSET ?` + + sqlUpdateDeliveryState = ` +UPDATE messages SET status=?, retry_count=?, last_attempt_at=? WHERE id=?` +) + +const schemaTemplate = ` CREATE TABLE IF NOT EXISTS messages ( id VARCHAR(255) PRIMARY KEY, version INT NOT NULL DEFAULT 1, @@ -188,3 +273,7 @@ CREATE TABLE IF NOT EXISTS messages ( INDEX idx_messages_input (input), INDEX idx_messages_created_at (created_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;` + +func buildSchemaSQL(tableName string) string { + return strings.ReplaceAll(schemaTemplate, "messages", tableName) +} diff --git a/internal/adapter/output/mariadb/repository_test.go b/internal/adapter/output/mariadb/repository_test.go index 984f3c4..8203cff 100644 --- a/internal/adapter/output/mariadb/repository_test.go +++ b/internal/adapter/output/mariadb/repository_test.go @@ -47,6 +47,61 @@ func newTestRepo(t *testing.T) *mariadbadapter.Repository { return repo } +func newTestRepoWithTableName(t *testing.T, tableName string) *mariadbadapter.Repository { + t.Helper() + if testing.Short() { + t.Skip("skipping MariaDB integration test (requires Docker)") + } + testcontainers.SkipIfProviderIsNotHealthy(t) + + ctx := context.Background() + ctr, err := tcmariadb.Run(ctx, "mariadb:11") + if err != nil { + t.Fatalf("start MariaDB container: %v", err) + } + t.Cleanup(func() { + if err := ctr.Terminate(ctx); err != nil { + t.Logf("terminate container: %v", err) + } + }) + + dsn, err := ctr.ConnectionString(ctx) + if err != nil { + t.Fatalf("get connection string: %v", err) + } + + repo, err := mariadbadapter.New(mariadbadapter.Config{DSN: dsn, TableName: tableName}) + if err != nil { + t.Fatalf("New() with tableName=%q error: %v", tableName, err) + } + t.Cleanup(func() { repo.Close() }) + return repo +} + +func TestRepository_CustomTableName(t *testing.T) { + repo := newTestRepoWithTableName(t, "custom_msgs") + ctx := context.Background() + + msg := domain.Message{ + ID: "01TEST", + Version: 1, + Input: "test-input", + Payload: domain.RawPayload(`{"key":"value"}`), + CreatedAt: time.Now().UTC().Truncate(time.Microsecond), + Status: domain.MessageStatusPending, + } + if err := repo.Save(ctx, msg); err != nil { + t.Fatalf("Save: %v", err) + } + got, err := repo.FindByID(ctx, msg.ID) + if err != nil { + t.Fatalf("FindByID: %v", err) + } + if got.ID != msg.ID { + t.Errorf("ID mismatch: got %q, want %q", got.ID, msg.ID) + } +} + func TestRepository_SaveAndFindByID(t *testing.T) { repo := newTestRepo(t) ctx := context.Background() diff --git a/internal/adapter/output/sqlite/repository.go b/internal/adapter/output/sqlite/repository.go index 43c5ce3..188f9c7 100644 --- a/internal/adapter/output/sqlite/repository.go +++ b/internal/adapter/output/sqlite/repository.go @@ -8,7 +8,6 @@ import ( "time" _ "modernc.org/sqlite" - "relaybox/internal/adapter/output/sqlite/db" output "relaybox/internal/application/port/output" "relaybox/internal/domain" ) @@ -17,33 +16,40 @@ import ( var _ output.MessageRepository = (*Repository)(nil) type Repository struct { - queries *db.Queries - sqlDB *sql.DB + sqlDB *sql.DB + tableName string } -func New(dsn string) (*Repository, error) { +func New(dsn, tableName string) (*Repository, error) { + if tableName == "" { + tableName = "messages" + } sqlDB, err := sql.Open("sqlite", dsn) if err != nil { return nil, fmt.Errorf("open sqlite: %w", err) } - if _, err := sqlDB.Exec(schemaSQL); err != nil { + if _, err := sqlDB.Exec(buildSchemaSQL(tableName)); err != nil { return nil, fmt.Errorf("apply schema: %w", err) } - return &Repository{queries: db.New(sqlDB), sqlDB: sqlDB}, nil + return &Repository{sqlDB: sqlDB, tableName: tableName}, nil } func (r *Repository) Close() error { return r.sqlDB.Close() } +func (r *Repository) buildSQL(query string) string { + return strings.ReplaceAll(query, "messages", r.tableName) +} + func (r *Repository) Save(ctx context.Context, m domain.Message) error { - err := r.queries.InsertMessage(ctx, db.InsertMessageParams{ - ID: m.ID, - Version: int64(m.Version), - Input: m.Input, - Payload: []byte(m.Payload), - CreatedAt: m.CreatedAt.UTC(), - Status: string(m.Status), - RetryCount: int64(m.RetryCount), - }) + _, err := r.sqlDB.ExecContext(ctx, r.buildSQL(sqlInsertMessage), + m.ID, + int64(m.Version), + m.Input, + []byte(m.Payload), + m.CreatedAt.UTC(), + string(m.Status), + int64(m.RetryCount), + ) if err != nil { return fmt.Errorf("save message: %w", err) } @@ -52,12 +58,12 @@ func (r *Repository) Save(ctx context.Context, m domain.Message) error { func (r *Repository) UpdateDeliveryState(ctx context.Context, id string, status domain.MessageStatus, retryCount int, lastAttemptAt time.Time) error { t := lastAttemptAt.UTC() - err := r.queries.UpdateDeliveryState(ctx, db.UpdateDeliveryStateParams{ - Status: string(status), - RetryCount: int64(retryCount), - LastAttemptAt: sql.NullTime{Time: t, Valid: true}, - ID: id, - }) + _, err := r.sqlDB.ExecContext(ctx, r.buildSQL(sqlUpdateDeliveryState), + string(status), + int64(retryCount), + sql.NullTime{Time: t, Valid: true}, + id, + ) if err != nil { return fmt.Errorf("update delivery state: %w", err) } @@ -65,23 +71,31 @@ func (r *Repository) UpdateDeliveryState(ctx context.Context, id string, status } func (r *Repository) FindByID(ctx context.Context, id string) (domain.Message, error) { - row, err := r.queries.GetMessageByID(ctx, id) + row := r.sqlDB.QueryRowContext(ctx, r.buildSQL(sqlGetMessageByID), id) + m, err := scanMessage(row) if err != nil { return domain.Message{}, fmt.Errorf("find message %q: %w", id, err) } - return toMessage(row), nil + return m, nil } func (r *Repository) FindByInput(ctx context.Context, inputID string, limit, offset int) ([]domain.Message, error) { - rows, err := r.queries.ListMessagesByInput(ctx, db.ListMessagesByInputParams{ - Input: inputID, Limit: int64(limit), Offset: int64(offset), - }) + rows, err := r.sqlDB.QueryContext(ctx, r.buildSQL(sqlListMessagesByInput), inputID, int64(limit), int64(offset)) if err != nil { return nil, fmt.Errorf("list messages: %w", err) } - messages := make([]domain.Message, 0, len(rows)) - for _, row := range rows { - messages = append(messages, toMessage(row)) + defer rows.Close() + + messages := make([]domain.Message, 0) + for rows.Next() { + m, err := scanMessageRow(rows) + if err != nil { + return nil, fmt.Errorf("scan message: %w", err) + } + messages = append(messages, m) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("rows error: %w", err) } return messages, nil } @@ -91,12 +105,12 @@ func (r *Repository) DeleteOlderThan(ctx context.Context, cutoff time.Time, stat var args []any if len(statuses) == 0 { - query = `DELETE FROM messages WHERE created_at < ?` + query = `DELETE FROM ` + r.tableName + ` WHERE created_at < ?` args = []any{cutoff.UTC()} } else { placeholders := strings.Repeat("?,", len(statuses)) placeholders = placeholders[:len(placeholders)-1] // trailing comma 제거 - query = `DELETE FROM messages WHERE created_at < ? AND status IN (` + placeholders + `)` + query = `DELETE FROM ` + r.tableName + ` WHERE created_at < ? AND status IN (` + placeholders + `)` args = make([]any, 0, 1+len(statuses)) args = append(args, cutoff.UTC()) for _, s := range statuses { @@ -115,24 +129,94 @@ func (r *Repository) DeleteOlderThan(ctx context.Context, cutoff time.Time, stat return n, nil } -func toMessage(row db.Message) domain.Message { - m := domain.Message{ - ID: row.ID, - Version: int(row.Version), - Input: row.Input, - Payload: domain.RawPayload(row.Payload), - CreatedAt: row.CreatedAt, - Status: domain.MessageStatus(row.Status), - RetryCount: int(row.RetryCount), - } - if row.LastAttemptAt.Valid { - t := row.LastAttemptAt.Time +type rowScanner interface { + Scan(dest ...any) error +} + +func scanMessage(row rowScanner) (domain.Message, error) { + var m domain.Message + var lastAttemptAt sql.NullTime + var createdAt time.Time + var status string + var version, retryCount int64 + var payload []byte + + err := row.Scan( + &m.ID, + &version, + &m.Input, + &payload, + &createdAt, + &status, + &retryCount, + &lastAttemptAt, + ) + if err != nil { + return domain.Message{}, err + } + m.Version = int(version) + m.Payload = domain.RawPayload(payload) + m.CreatedAt = createdAt + m.Status = domain.MessageStatus(status) + m.RetryCount = int(retryCount) + if lastAttemptAt.Valid { + t := lastAttemptAt.Time m.LastAttemptAt = &t } - return m + return m, nil } -const schemaSQL = ` +func scanMessageRow(rows *sql.Rows) (domain.Message, error) { + var m domain.Message + var lastAttemptAt sql.NullTime + var createdAt time.Time + var status string + var version, retryCount int64 + var payload []byte + + err := rows.Scan( + &m.ID, + &version, + &m.Input, + &payload, + &createdAt, + &status, + &retryCount, + &lastAttemptAt, + ) + if err != nil { + return domain.Message{}, err + } + m.Version = int(version) + m.Payload = domain.RawPayload(payload) + m.CreatedAt = createdAt + m.Status = domain.MessageStatus(status) + m.RetryCount = int(retryCount) + if lastAttemptAt.Valid { + t := lastAttemptAt.Time + m.LastAttemptAt = &t + } + return m, nil +} + +const ( + sqlInsertMessage = ` +INSERT INTO messages (id, version, input, payload, created_at, status, retry_count) +VALUES (?, ?, ?, ?, ?, ?, ?)` + + sqlGetMessageByID = ` +SELECT id, version, input, payload, created_at, status, retry_count, last_attempt_at +FROM messages WHERE id=?` + + sqlListMessagesByInput = ` +SELECT id, version, input, payload, created_at, status, retry_count, last_attempt_at +FROM messages WHERE input=? ORDER BY created_at DESC LIMIT ? OFFSET ?` + + sqlUpdateDeliveryState = ` +UPDATE messages SET status=?, retry_count=?, last_attempt_at=? WHERE id=?` +) + +const schemaTemplate = ` CREATE TABLE IF NOT EXISTS messages ( id TEXT PRIMARY KEY, version INTEGER NOT NULL DEFAULT 1, @@ -144,3 +228,7 @@ CREATE TABLE IF NOT EXISTS messages ( last_attempt_at DATETIME ); CREATE INDEX IF NOT EXISTS idx_messages_created_at ON messages(created_at);` + +func buildSchemaSQL(tableName string) string { + return strings.ReplaceAll(schemaTemplate, "messages", tableName) +} diff --git a/internal/adapter/output/sqlite/repository_test.go b/internal/adapter/output/sqlite/repository_test.go index d40835a..44e069d 100644 --- a/internal/adapter/output/sqlite/repository_test.go +++ b/internal/adapter/output/sqlite/repository_test.go @@ -2,6 +2,7 @@ package sqlite_test import ( "context" + "path/filepath" "testing" "time" @@ -17,7 +18,7 @@ var _ interface { func newTestRepo(t *testing.T) *sqliteadapter.Repository { t.Helper() - repo, err := sqliteadapter.New(":memory:") + repo, err := sqliteadapter.New(":memory:", "") if err != nil { t.Fatalf("New() error: %v", err) } @@ -25,6 +26,34 @@ func newTestRepo(t *testing.T) *sqliteadapter.Repository { return repo } +func TestRepository_CustomTableName(t *testing.T) { + dir := t.TempDir() + repo, err := sqliteadapter.New(filepath.Join(dir, "test.db"), "custom_msgs") + if err != nil { + t.Fatalf("New with custom table: %v", err) + } + defer repo.Close() + + msg := domain.Message{ + ID: "01TEST", + Version: 1, + Input: "test-input", + Payload: domain.RawPayload(`{"key":"value"}`), + CreatedAt: time.Now().UTC().Truncate(time.Second), + Status: domain.MessageStatusPending, + } + if err := repo.Save(context.Background(), msg); err != nil { + t.Fatalf("Save: %v", err) + } + got, err := repo.FindByID(context.Background(), msg.ID) + if err != nil { + t.Fatalf("FindByID: %v", err) + } + if got.ID != msg.ID { + t.Errorf("ID mismatch: got %q, want %q", got.ID, msg.ID) + } +} + func TestRepository_SaveAndFindByID(t *testing.T) { repo := newTestRepo(t) ctx := context.Background() diff --git a/internal/adapter/output/webhook/sender.go b/internal/adapter/output/webhook/sender.go index 1e383e7..37333b9 100644 --- a/internal/adapter/output/webhook/sender.go +++ b/internal/adapter/output/webhook/sender.go @@ -6,6 +6,7 @@ import ( "crypto/tls" "fmt" "io" + "log/slog" "net/http" "time" @@ -44,12 +45,26 @@ func (s *Sender) Send(ctx context.Context, out domain.Output, payload []byte) er if out.Secret != "" { req.Header.Set("Authorization", "Bearer "+out.Secret) } + start := time.Now() resp, err := client.Do(req) if err != nil { return fmt.Errorf("send: %w", err) } defer resp.Body.Close() - _, _ = io.Copy(io.Discard, resp.Body) + respBody, _ := io.ReadAll(resp.Body) + slog.Info("webhook sent", + "output", out.ID, + "url", out.URL, + "statusCode", resp.StatusCode, + "elapsed", time.Since(start), + ) + slog.Debug("webhook detail", + "output", out.ID, + "requestBody", string(payload), + "responseBody", string(respBody), + "requestHeaders", req.Header, + "responseHeaders", resp.Header, + ) if resp.StatusCode >= 400 { return fmt.Errorf("webhook returned %d", resp.StatusCode) } diff --git a/internal/adapter/output/webhook/sender_test.go b/internal/adapter/output/webhook/sender_test.go index 7b12ad3..709fd85 100644 --- a/internal/adapter/output/webhook/sender_test.go +++ b/internal/adapter/output/webhook/sender_test.go @@ -3,6 +3,7 @@ package webhook_test import ( "context" "io" + "log/slog" "net/http" "net/http/httptest" "testing" @@ -11,6 +12,7 @@ import ( "relaybox/internal/adapter/output/webhook" "relaybox/internal/application/port/output" "relaybox/internal/domain" + "relaybox/internal/testutil" ) // compile-time interface check @@ -64,6 +66,45 @@ func TestSender_Send(t *testing.T) { } } +func TestSender_LogsInfoOnSuccess(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + h := &testutil.CaptureHandler{} + orig := slog.Default() + slog.SetDefault(slog.New(h)) + defer slog.SetDefault(orig) + + sender := webhook.NewSender() + out := domain.Output{ID: "test-output", URL: srv.URL} + + if err := sender.Send(context.Background(), out, []byte(`{"test":true}`)); err != nil { + t.Fatalf("Send() error: %v", err) + } + + var found bool + for _, rec := range h.Records() { + if rec.Level == slog.LevelInfo && rec.Msg == "webhook sent" { + if _, ok := rec.Attrs["output"]; !ok { + t.Error("log record missing key: output") + } + if _, ok := rec.Attrs["statusCode"]; !ok { + t.Error("log record missing key: statusCode") + } + if _, ok := rec.Attrs["elapsed"]; !ok { + t.Error("log record missing key: elapsed") + } + found = true + break + } + } + if !found { + t.Error(`expected INFO "webhook sent" log record not found`) + } +} + func TestRegistry_Get(t *testing.T) { reg := webhook.NewRegistry(map[domain.OutputType]output.OutputSender{ domain.OutputTypeWebhook: webhook.NewSender(), diff --git a/internal/application/service/relay_worker.go b/internal/application/service/relay_worker.go index 2ab3326..c953597 100644 --- a/internal/application/service/relay_worker.go +++ b/internal/application/service/relay_worker.go @@ -77,6 +77,8 @@ func (w *RelayWorker) processOne(ctx context.Context) error { defer w.cfg.Hooks.OnProcessed() } + slog.Info("processing message", "messageID", msg.ID, "input", msg.Input) + inputEngine, entries, err := w.ruleReader.GetRules(ctx, msg.Input) if err != nil { _ = nack() @@ -173,6 +175,7 @@ func (w *RelayWorker) processOne(ctx context.Context) error { success = false continue } + slog.Debug("delivering to output", "messageID", msg.ID, "output", out.ID, "payloadSize", len(payload)) if err := w.deliver(ctx, out, payload); err != nil { slog.Warn("delivery failed", "messageID", msg.ID, "output", out.ID, "err", err) success = false @@ -182,6 +185,7 @@ func (w *RelayWorker) processOne(ctx context.Context) error { now := time.Now().UTC() if success { + slog.Info("message delivered", "messageID", msg.ID, "input", msg.Input) if msg.Status.CanTransitionTo(domain.MessageStatusDelivered) { if err := w.repo.UpdateDeliveryState(ctx, msg.ID, domain.MessageStatusDelivered, msg.RetryCount, now); err != nil { slog.Error("failed to update delivery state to delivered", "messageID", msg.ID, "err", err) @@ -193,6 +197,7 @@ func (w *RelayWorker) processOne(ctx context.Context) error { slog.Warn("ack failed", "messageID", msg.ID, "err", err) } } else { + slog.Info("message failed", "messageID", msg.ID, "input", msg.Input, "retryCount", msg.RetryCount+1) if msg.Status.CanTransitionTo(domain.MessageStatusFailed) { if err := w.repo.UpdateDeliveryState(ctx, msg.ID, domain.MessageStatusFailed, msg.RetryCount+1, now); err != nil { slog.Error("failed to update delivery state to failed", "messageID", msg.ID, "err", err) diff --git a/internal/application/service/relay_worker_test.go b/internal/application/service/relay_worker_test.go index beff570..3fc94ca 100644 --- a/internal/application/service/relay_worker_test.go +++ b/internal/application/service/relay_worker_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "log/slog" "sync" "sync/atomic" "testing" @@ -13,6 +14,7 @@ import ( "relaybox/internal/application/port/output" "relaybox/internal/application/service" "relaybox/internal/domain" + "relaybox/internal/testutil" ) type mockMessageQueue struct { @@ -625,6 +627,58 @@ func TestRelayWorker_Mapping_EnrichesData(t *testing.T) { } } +func TestRelayWorker_LogsInfoOnProcessing(t *testing.T) { + h := &testutil.CaptureHandler{} + orig := slog.Default() + slog.SetDefault(slog.New(h)) + defer slog.SetDefault(orig) + + msgObj := domain.Message{ + ID: "log-test", Input: "beszel", + Payload: domain.RawPayload(`{"host":"server1"}`), + Status: domain.MessageStatusPending, Version: 1, + } + queue := &mockMessageQueue{messages: []domain.Message{msgObj}} + repo := &mockRepo{saveFn: func(_ context.Context, _ domain.Message) error { return nil }} + sender := &mockSender{} + ruleReader := &mockRuleReader{ + engine: "CEL", + entries: []domain.RuleEntry{{ + Rule: domain.Rule{}, + Outputs: []domain.Output{{ID: "c1", Type: domain.OutputTypeWebhook, Engine: "CEL"}}, + }}, + } + registry := &mockRegistry{sender: sender} + + processed, onProcessed := processedChan() + cfg := service.DefaultRelayWorkerConfig() + cfg.Hooks.OnProcessed = onProcessed + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + worker := service.NewRelayWorker(queue, repo, ruleReader, registry, newExprRegistry(), cfg) + worker.Start(ctx, 1) + waitForProcessed(t, processed) + + hasProcessingLog := false + hasDeliveredLog := false + for _, rec := range h.Records() { + if rec.Level == slog.LevelInfo && rec.Msg == "processing message" { + hasProcessingLog = true + } + if rec.Level == slog.LevelInfo && rec.Msg == "message delivered" { + hasDeliveredLog = true + } + } + if !hasProcessingLog { + t.Error(`expected INFO "processing message" log record not found`) + } + if !hasDeliveredLog { + t.Error(`expected INFO "message delivered" log record not found`) + } +} + // mockCountingSenderError counts Send calls and always returns an error. type mockCountingSenderError struct { count atomic.Int32 diff --git a/internal/config/config.go b/internal/config/config.go index 5c2a347..7d9bed0 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -4,6 +4,7 @@ import ( "fmt" "log/slog" "net" + "regexp" "strings" "time" @@ -82,6 +83,7 @@ type StorageConfig struct { MaxOpenConns int `mapstructure:"maxOpenConns"` MaxIdleConns int `mapstructure:"maxIdleConns"` ConnMaxLifetime string `mapstructure:"connMaxLifetime"` // Go duration, e.g. "5m" + TableName string `mapstructure:"tableName"` // 메시지 테이블 이름 (기본값: messages) Rotation RotationConfig `mapstructure:"rotation"` } @@ -152,6 +154,7 @@ func setDefaults(v *viper.Viper) { v.SetDefault("worker.defaultRetryCount", 3) v.SetDefault("worker.defaultRetryDelay", "1s") v.SetDefault("worker.pollBackoff", "500ms") + v.SetDefault("storage.tableName", "messages") v.SetDefault("storage.rotation.enabled", false) v.SetDefault("storage.rotation.retention", "720h") v.SetDefault("storage.rotation.interval", "1h") @@ -169,11 +172,15 @@ func unmarshalAndValidate(v *viper.Viper) (*Config, error) { } var validEngines = map[string]struct{}{"CEL": {}, "EXPR": {}} +var validTableName = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]{0,63}$`) func validateConfig(cfg *Config) error { if strings.ToUpper(cfg.Storage.Type) == "MARIADB" && cfg.Storage.DSN == "" { return fmt.Errorf("storage.dsn is required when storage.type is MARIADB") } + if cfg.Storage.TableName != "" && !validTableName.MatchString(cfg.Storage.TableName) { + return fmt.Errorf("storage.tableName: invalid name %q (must match ^[a-zA-Z_][a-zA-Z0-9_]{0,63}$)", cfg.Storage.TableName) + } if cfg.Storage.ConnMaxLifetime != "" { if _, err := time.ParseDuration(cfg.Storage.ConnMaxLifetime); err != nil { return fmt.Errorf("storage.connMaxLifetime: invalid duration %q: %w", cfg.Storage.ConnMaxLifetime, err) diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 9bca90b..633abfe 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -489,3 +489,37 @@ storage: t.Errorf("error should mention rotation.interval, got: %v", err) } } + +func TestValidateConfig_TableName(t *testing.T) { + tests := []struct { + name string + tableName string + wantErr bool + }{ + {"default messages", "messages", false}, + {"with underscore", "relay_messages", false}, + {"with number suffix", "my_table_1", false}, + {"starts with underscore", "_table", false}, + {"starts with number", "1table", true}, + {"contains space", "my table", true}, + {"sql injection", "messages; DROP TABLE messages;--", true}, + {"too long 65 chars", "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", true}, // 65자 + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + yaml := rotationBaseYAML + ` +storage: + type: SQLITE + path: ./data/test.db + tableName: ` + tc.tableName + ` +` + _, err := config.Load(writeConfig(t, yaml)) + if tc.wantErr && err == nil { + t.Errorf("expected error for tableName=%q, got nil", tc.tableName) + } + if !tc.wantErr && err != nil { + t.Errorf("unexpected error for tableName=%q: %v", tc.tableName, err) + } + }) + } +} diff --git a/internal/testutil/slog.go b/internal/testutil/slog.go new file mode 100644 index 0000000..52e0540 --- /dev/null +++ b/internal/testutil/slog.go @@ -0,0 +1,47 @@ +// Package testutil provides shared test helpers. +package testutil + +import ( + "context" + "log/slog" + "sync" +) + +// LogRecord holds a captured slog record. +type LogRecord struct { + Level slog.Level + Msg string + Attrs map[string]any +} + +// CaptureHandler is a minimal slog.Handler that collects log records. +type CaptureHandler struct { + mu sync.Mutex + records []LogRecord +} + +func (h *CaptureHandler) Enabled(_ context.Context, _ slog.Level) bool { return true } + +func (h *CaptureHandler) Handle(_ context.Context, r slog.Record) error { + attrs := make(map[string]any) + r.Attrs(func(a slog.Attr) bool { + attrs[a.Key] = a.Value.Any() + return true + }) + h.mu.Lock() + h.records = append(h.records, LogRecord{Level: r.Level, Msg: r.Message, Attrs: attrs}) + h.mu.Unlock() + return nil +} + +func (h *CaptureHandler) WithAttrs(_ []slog.Attr) slog.Handler { return h } +func (h *CaptureHandler) WithGroup(_ string) slog.Handler { return h } + +// Records returns a copy of all captured records. +func (h *CaptureHandler) Records() []LogRecord { + h.mu.Lock() + defer h.mu.Unlock() + out := make([]LogRecord, len(h.records)) + copy(out, h.records) + return out +} diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go index 6804677..fee9421 100644 --- a/test/e2e/e2e_test.go +++ b/test/e2e/e2e_test.go @@ -81,7 +81,7 @@ func TestE2E_PostMessage_Returns201(t *testing.T) { Queue: cfgpkg.QueueConfig{WorkerCount: 1}, } - repo, _ := sqliteadapter.New(":memory:") + repo, _ := sqliteadapter.New(":memory:", "") defer repo.Close() queue, _ := filequeue.New(t.TempDir()) sender := webhookadapter.NewSender() @@ -165,7 +165,7 @@ func TestE2E_ListMessages(t *testing.T) { Outputs: []cfgpkg.OutputConfig{{ID: "wh1", Type: "WEBHOOK", Engine: "CEL", URL: "http://localhost:9999"}}, Queue: cfgpkg.QueueConfig{WorkerCount: 1}, } - repo, _ := sqliteadapter.New(":memory:") + repo, _ := sqliteadapter.New(":memory:", "") defer repo.Close() queue, _ := filequeue.New(t.TempDir()) msgSvc := service.NewMessageService(repo, queue, nil, nil) @@ -220,7 +220,7 @@ func TestE2E_ConfigEndpoints(t *testing.T) { {ID: "wh1", Type: "WEBHOOK", Engine: "CEL", URL: "http://example.com/hook", RetryCount: 3, RetryDelayMs: 100}, }, } - repo, _ := sqliteadapter.New(":memory:") + repo, _ := sqliteadapter.New(":memory:", "") defer repo.Close() queue, _ := filequeue.New(t.TempDir()) msgSvc := service.NewMessageService(repo, queue, nil, nil) @@ -302,7 +302,7 @@ func TestE2E_ConfigEndpoints(t *testing.T) { } func TestE2E_Healthz(t *testing.T) { - repo, _ := sqliteadapter.New(":memory:") + repo, _ := sqliteadapter.New(":memory:", "") defer repo.Close() queue, _ := filequeue.New(t.TempDir()) msgSvc := service.NewMessageService(repo, queue, nil, nil)