hyapp-server/services/notice-service/internal/modules/roomnotice/mysql_integration_test.go

package roomnotice

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"os"
	"strings"
	"testing"
	"time"

	"google.golang.org/protobuf/proto"
	roomeventsv1 "hyapp.local/api/proto/events/room/v1"

	_ "github.com/go-sql-driver/mysql"
)

func TestMySQLRepositoryProcessesRealRoomKickOutbox(t *testing.T) {
	dsn := strings.TrimSpace(os.Getenv("NOTICE_REAL_MYSQL_DSN"))
	if dsn == "" {
		t.Skip("set NOTICE_REAL_MYSQL_DSN to run real MySQL notice validation")
	}
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		t.Fatalf("open mysql: %v", err)
	}
	defer db.Close()
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if err := db.PingContext(ctx); err != nil {
		t.Fatalf("ping mysql: %v", err)
	}

	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
	roomDatabase := strings.TrimSpace(os.Getenv("NOTICE_REAL_ROOM_DATABASE"))
	if roomDatabase == "" {
		roomDatabase = "hyapp_room_notice_test_" + suffix
		if _, err := db.ExecContext(ctx, fmt.Sprintf(`CREATE DATABASE %s DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci`, quoteDB(roomDatabase))); err != nil {
			t.Fatalf("create room test database: %v", err)
		}
		t.Cleanup(func() {
			_, _ = db.ExecContext(context.Background(), fmt.Sprintf(`DROP DATABASE %s`, quoteDB(roomDatabase)))
		})
	}
	if err := ensureRoomOutboxTable(ctx, db, roomDatabase); err != nil {
		t.Fatalf("ensure room_outbox table: %v", err)
	}

	repository, err := NewMySQLRepository(db, roomDatabase)
	if err != nil {
		t.Fatalf("NewMySQLRepository: %v", err)
	}
	appCode := "notice_real"
	roomID := "notice_real_room_" + suffix
	eventID := "notice_real_room_evt_" + suffix
	actorUserID := int64(990000001)
	targetUserID := int64(990000002)
	nowMS := time.Now().UnixMilli()

	body, err := proto.Marshal(&roomeventsv1.RoomUserKicked{
		ActorUserId:  actorUserID,
		TargetUserId: targetUserID,
	})
	if err != nil {
		t.Fatalf("marshal body: %v", err)
	}
	envelope, err := proto.Marshal(&roomeventsv1.EventEnvelope{
		EventId:      eventID,
		RoomId:       roomID,
		EventType:    "RoomUserKicked",
		RoomVersion:  12,
		OccurredAtMs: nowMS,
		AppCode:      appCode,
		Body:         body,
	})
	if err != nil {
		t.Fatalf("marshal envelope: %v", err)
	}

	if _, err := db.ExecContext(ctx, fmt.Sprintf(`
		INSERT INTO %s.room_outbox (
		  app_code, event_id, event_type, room_id, status, worker_id, lock_until_ms,
		  envelope, created_at_ms, retry_count, next_retry_at_ms, last_error, updated_at_ms
		) VALUES (?, ?, 'RoomUserKicked', ?, 'delivered', '', 0, ?, ?, 0, 0, '', ?)`,
		quoteDB(roomDatabase),
	), appCode, eventID, roomID, envelope, nowMS, nowMS); err != nil {
		t.Fatalf("insert room_outbox: %v", err)
	}
	if !keepRealRows() {
		t.Cleanup(func() {
			_, _ = db.ExecContext(context.Background(), `DELETE FROM notice_delivery_events WHERE app_code = ? AND source_event_id = ?`, appCode, eventID)
			_, _ = db.ExecContext(context.Background(), fmt.Sprintf(`DELETE FROM %s.room_outbox WHERE app_code = ? AND event_id = ?`, quoteDB(roomDatabase)), appCode, eventID)
		})
	}

	publisher := &fakePublisher{}
	service := New(Config{NodeID: "notice-real-test"}, repository, publisher)
	processed, err := service.ProcessRoomKickNotices(ctx, RoomNoticeWorkerOptions{
		WorkerID:       "notice-real-room-test-worker",
		BatchSize:      1,
		LockTTL:        30 * time.Second,
		PublishTimeout: time.Second,
		MaxRetryCount:  3,
	})
	if err != nil {
		t.Fatalf("ProcessRoomKickNotices: %v", err)
	}
	if processed != 1 || len(publisher.messages) != 1 {
		t.Fatalf("unexpected process result: processed=%d messages=%d", processed, len(publisher.messages))
	}
	if publisher.messages[0].ToAccount != "990000002" || publisher.messages[0].Ext != "room_notice" {
		t.Fatalf("unexpected private message: %+v", publisher.messages[0])
	}

	var noticeStatus string
	var deliveredAtMS int64
	var storedPayload string
	if err := db.QueryRowContext(ctx, `
		SELECT status, delivered_at_ms, COALESCE(CAST(payload_json AS CHAR), '{}')
		FROM notice_delivery_events
		WHERE source_name = 'room_outbox' AND app_code = ? AND source_event_id = ? AND channel = 'tencent_im_c2c'`,
		appCode, eventID,
	).Scan(&noticeStatus, &deliveredAtMS, &storedPayload); err != nil {
		t.Fatalf("query notice_delivery_events: %v", err)
	}
	if noticeStatus != "delivered" || deliveredAtMS <= 0 {
		t.Fatalf("unexpected delivery marker: status=%s delivered_at_ms=%d payload=%s", noticeStatus, deliveredAtMS, storedPayload)
	}
	var noticePayload map[string]any
	if err := json.Unmarshal([]byte(storedPayload), &noticePayload); err != nil {
		t.Fatalf("decode stored notice payload: %v", err)
	}
	if noticePayload["event_id"] != eventID || noticePayload["room_id"] != roomID || noticePayload["event_type"] != "room_user_kicked" {
		t.Fatalf("unexpected stored notice payload: %+v", noticePayload)
	}
	if noticePayload["target_user_id"].(float64) != float64(targetUserID) || noticePayload["actor_user_id"].(float64) != float64(actorUserID) {
		t.Fatalf("unexpected stored user payload: %+v", noticePayload)
	}
	t.Logf("real room notice validation event_id=%s status=%s delivered_at_ms=%d", eventID, noticeStatus, deliveredAtMS)
}

func keepRealRows() bool {
	value := strings.ToLower(strings.TrimSpace(os.Getenv("NOTICE_REAL_MYSQL_KEEP_ROWS")))
	return value == "1" || value == "true" || value == "yes"
}

func quoteDB(value string) string {
	return "`" + strings.ReplaceAll(value, "`", "``") + "`"
}

func ensureRoomOutboxTable(ctx context.Context, db *sql.DB, roomDatabase string) error {
	_, err := db.ExecContext(ctx, fmt.Sprintf(`
		CREATE TABLE IF NOT EXISTS %s.room_outbox (
		  app_code VARCHAR(32) NOT NULL,
		  event_id VARCHAR(128) NOT NULL,
		  event_type VARCHAR(64) NOT NULL,
		  room_id VARCHAR(64) NOT NULL,
		  status VARCHAR(32) NOT NULL,
		  worker_id VARCHAR(128) NOT NULL DEFAULT '',
		  lock_until_ms BIGINT NULL,
		  envelope LONGBLOB NOT NULL,
		  created_at_ms BIGINT NOT NULL,
		  retry_count INT NOT NULL DEFAULT 0,
		  next_retry_at_ms BIGINT NOT NULL DEFAULT 0,
		  last_error TEXT NULL,
		  updated_at_ms BIGINT NOT NULL,
		  PRIMARY KEY (app_code, event_id),
		  KEY idx_room_outbox_notice (app_code, event_type, created_at_ms)
		) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
		quoteDB(roomDatabase),
	))
	return err
}