2026-05-08 18:48:29 +08:00

472 lines
18 KiB
Go

// Package mictime stores user-scoped microphone online duration read models in MySQL.
package mictime
import (
"context"
"database/sql"
"errors"
"strings"
"time"
mysqldriver "github.com/go-sql-driver/mysql"
"hyapp/pkg/appcode"
"hyapp/pkg/xerr"
mictimedomain "hyapp/services/user-service/internal/domain/mictime"
)
// Repository owns mic session state and user duration aggregates inside user-service MySQL.
type Repository struct {
db *sql.DB
}
// New creates a mic duration repository backed by the shared user-service connection pool.
func New(db *sql.DB) *Repository {
return &Repository{db: db}
}
// ProcessRoomMicEvent idempotently applies one RoomMicChanged fact.
func (r *Repository) ProcessRoomMicEvent(ctx context.Context, event mictimedomain.RoomMicEvent) (mictimedomain.ApplyResult, error) {
if r == nil || r.db == nil {
return mictimedomain.ApplyResult{}, xerr.New(xerr.Unavailable, "mic time repository is not configured")
}
event = normalizeRoomMicEvent(event)
if event.EventID == "" || event.EventType == "" {
return mictimedomain.ApplyResult{}, xerr.New(xerr.InvalidArgument, "room mic event identity is required")
}
if event.OccurredAtMs <= 0 {
event.OccurredAtMs = time.Now().UnixMilli()
}
tx, err := r.db.BeginTx(ctx, nil)
if err != nil {
return mictimedomain.ApplyResult{}, err
}
defer tx.Rollback()
if err := insertEventConsumption(ctx, tx, event); err != nil {
if isDuplicate(err) {
return mictimedomain.ApplyResult{Skipped: true, Reason: "already_consumed"}, nil
}
return mictimedomain.ApplyResult{}, err
}
if event.EventType != mictimedomain.RoomMicChangedEventType {
if err := finishEventConsumption(ctx, tx, event, "skipped", "not_room_mic_changed"); err != nil {
return mictimedomain.ApplyResult{}, err
}
return mictimedomain.ApplyResult{Skipped: true, Reason: "not_room_mic_changed"}, tx.Commit()
}
if event.MicSessionID == "" || event.TargetUserID <= 0 {
if err := finishEventConsumption(ctx, tx, event, "skipped", "incomplete_mic_event"); err != nil {
return mictimedomain.ApplyResult{}, err
}
return mictimedomain.ApplyResult{Skipped: true, Reason: "incomplete_mic_event"}, tx.Commit()
}
var result mictimedomain.ApplyResult
switch event.Action {
case mictimedomain.ActionUp:
result, err = applyUp(ctx, tx, event)
case mictimedomain.ActionPublishConfirmed:
result, err = applyPublishConfirmed(ctx, tx, event)
case mictimedomain.ActionChange:
result, err = applyChange(ctx, tx, event)
case mictimedomain.ActionDown:
result, err = applyDown(ctx, tx, event)
default:
result = mictimedomain.ApplyResult{Skipped: true, Reason: "unsupported_mic_action"}
}
if err != nil {
return mictimedomain.ApplyResult{}, err
}
if result.Reason == "" && result.Skipped {
result.Reason = "skipped"
}
status := "consumed"
if result.Skipped {
status = "skipped"
}
if err := finishEventConsumption(ctx, tx, event, status, result.Reason); err != nil {
return mictimedomain.ApplyResult{}, err
}
if err := tx.Commit(); err != nil {
return mictimedomain.ApplyResult{}, err
}
if !result.Skipped {
result.Consumed = true
}
return result, nil
}
// GetLifetimeStats returns the current user-wide mic duration aggregate.
func (r *Repository) GetLifetimeStats(ctx context.Context, appCode string, userID int64) (mictimedomain.LifetimeStats, error) {
if r == nil || r.db == nil {
return mictimedomain.LifetimeStats{}, xerr.New(xerr.Unavailable, "mic time repository is not configured")
}
if userID <= 0 {
return mictimedomain.LifetimeStats{}, xerr.New(xerr.InvalidArgument, "user_id is required")
}
stats := mictimedomain.LifetimeStats{AppCode: appcode.Normalize(appCode), UserID: userID}
err := r.db.QueryRowContext(ctx, `
SELECT seat_occupied_ms, mic_online_ms, session_count, first_mic_at_ms, last_mic_at_ms, updated_at_ms
FROM user_mic_lifetime_stats
WHERE app_code = ? AND user_id = ?
`, stats.AppCode, userID).Scan(&stats.SeatOccupiedMs, &stats.MicOnlineMs, &stats.SessionCount, &stats.FirstMicAtMs, &stats.LastMicAtMs, &stats.UpdatedAtMs)
if err == sql.ErrNoRows {
return stats, nil
}
if err != nil {
return mictimedomain.LifetimeStats{}, err
}
return stats, nil
}
func applyUp(ctx context.Context, tx *sql.Tx, event mictimedomain.RoomMicEvent) (mictimedomain.ApplyResult, error) {
seatNo := event.ToSeat
if seatNo <= 0 {
seatNo = event.FromSeat
}
if seatNo <= 0 {
return mictimedomain.ApplyResult{Skipped: true, Reason: "missing_seat_no"}, nil
}
_, err := tx.ExecContext(ctx, `
INSERT INTO user_mic_sessions (
app_code, mic_session_id, user_id, room_id, first_seat_no, current_seat_no, status,
seat_started_at_ms, publishing_started_at_ms, ended_at_ms, end_reason,
seat_occupied_ms, mic_online_ms, opened_event_id, publish_event_id, closed_event_id,
last_event_id, last_room_version, created_at_ms, updated_at_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, '', 0, 0, ?, '', '', ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
current_seat_no = CASE WHEN status = ? THEN current_seat_no ELSE VALUES(current_seat_no) END,
last_event_id = VALUES(last_event_id),
last_room_version = GREATEST(last_room_version, VALUES(last_room_version)),
updated_at_ms = VALUES(updated_at_ms)
`, event.AppCode, event.MicSessionID, event.TargetUserID, event.RoomID, seatNo, seatNo, mictimedomain.SessionStatusPendingPublish,
event.OccurredAtMs, event.EventID, event.EventID, event.RoomVersion, event.OccurredAtMs, event.OccurredAtMs, mictimedomain.SessionStatusEnded)
if err != nil {
return mictimedomain.ApplyResult{}, err
}
return mictimedomain.ApplyResult{Consumed: true}, nil
}
func applyPublishConfirmed(ctx context.Context, tx *sql.Tx, event mictimedomain.RoomMicEvent) (mictimedomain.ApplyResult, error) {
session, exists, err := sessionForUpdate(ctx, tx, event.AppCode, event.MicSessionID)
if err != nil || !exists {
if err != nil {
return mictimedomain.ApplyResult{}, err
}
return mictimedomain.ApplyResult{Skipped: true, Reason: "session_not_found"}, nil
}
if session.UserID != event.TargetUserID {
return mictimedomain.ApplyResult{Skipped: true, Reason: "target_user_mismatch"}, nil
}
if session.Status == mictimedomain.SessionStatusEnded {
return mictimedomain.ApplyResult{Skipped: true, Reason: "session_already_ended"}, nil
}
publishAt := event.PublishEventTimeMs
if publishAt <= 0 {
publishAt = event.OccurredAtMs
}
if publishAt < session.SeatStartedAtMs {
// RTC/client clocks can arrive slightly earlier than server MicUp time; clamp to avoid negative online windows.
publishAt = session.SeatStartedAtMs
}
if session.PublishingStartedAtMs.Valid && publishAt <= session.PublishingStartedAtMs.Int64 {
if err := updateSessionLastEvent(ctx, tx, event); err != nil {
return mictimedomain.ApplyResult{}, err
}
return mictimedomain.ApplyResult{Skipped: true, Reason: "publish_already_confirmed"}, nil
}
_, err = tx.ExecContext(ctx, `
UPDATE user_mic_sessions
SET status = ?, publishing_started_at_ms = ?, publish_event_id = ?, last_event_id = ?,
last_room_version = GREATEST(last_room_version, ?), updated_at_ms = ?
WHERE app_code = ? AND mic_session_id = ? AND status <> ?
`, mictimedomain.SessionStatusPublishing, publishAt, event.EventID, event.EventID, event.RoomVersion, event.OccurredAtMs,
event.AppCode, event.MicSessionID, mictimedomain.SessionStatusEnded)
if err != nil {
return mictimedomain.ApplyResult{}, err
}
return mictimedomain.ApplyResult{Consumed: true}, nil
}
func applyChange(ctx context.Context, tx *sql.Tx, event mictimedomain.RoomMicEvent) (mictimedomain.ApplyResult, error) {
session, exists, err := sessionForUpdate(ctx, tx, event.AppCode, event.MicSessionID)
if err != nil || !exists {
if err != nil {
return mictimedomain.ApplyResult{}, err
}
return mictimedomain.ApplyResult{Skipped: true, Reason: "session_not_found"}, nil
}
if session.UserID != event.TargetUserID {
return mictimedomain.ApplyResult{Skipped: true, Reason: "target_user_mismatch"}, nil
}
if session.Status == mictimedomain.SessionStatusEnded {
return mictimedomain.ApplyResult{Skipped: true, Reason: "session_already_ended"}, nil
}
if event.ToSeat <= 0 {
return mictimedomain.ApplyResult{Skipped: true, Reason: "missing_target_seat"}, nil
}
_, err = tx.ExecContext(ctx, `
UPDATE user_mic_sessions
SET current_seat_no = ?, last_event_id = ?, last_room_version = GREATEST(last_room_version, ?), updated_at_ms = ?
WHERE app_code = ? AND mic_session_id = ? AND status <> ?
`, event.ToSeat, event.EventID, event.RoomVersion, event.OccurredAtMs, event.AppCode, event.MicSessionID, mictimedomain.SessionStatusEnded)
if err != nil {
return mictimedomain.ApplyResult{}, err
}
return mictimedomain.ApplyResult{Consumed: true}, nil
}
func applyDown(ctx context.Context, tx *sql.Tx, event mictimedomain.RoomMicEvent) (mictimedomain.ApplyResult, error) {
session, exists, err := sessionForUpdate(ctx, tx, event.AppCode, event.MicSessionID)
if err != nil || !exists {
if err != nil {
return mictimedomain.ApplyResult{}, err
}
return mictimedomain.ApplyResult{Skipped: true, Reason: "session_not_found"}, nil
}
if session.UserID != event.TargetUserID {
return mictimedomain.ApplyResult{Skipped: true, Reason: "target_user_mismatch"}, nil
}
if session.Status == mictimedomain.SessionStatusEnded {
return mictimedomain.ApplyResult{Skipped: true, Reason: "session_already_ended"}, nil
}
endAt := event.OccurredAtMs
if endAt < session.SeatStartedAtMs {
endAt = session.SeatStartedAtMs
}
seatOccupiedMs := maxInt64(0, endAt-session.SeatStartedAtMs)
micOnlineMs := int64(0)
if session.PublishingStartedAtMs.Valid {
publishStart := session.PublishingStartedAtMs.Int64
if publishStart < session.SeatStartedAtMs {
publishStart = session.SeatStartedAtMs
}
micOnlineMs = maxInt64(0, endAt-publishStart)
}
endReason := strings.TrimSpace(event.Reason)
if endReason == "" {
endReason = "down"
}
_, err = tx.ExecContext(ctx, `
UPDATE user_mic_sessions
SET status = ?, ended_at_ms = ?, end_reason = ?, seat_occupied_ms = ?, mic_online_ms = ?,
closed_event_id = ?, last_event_id = ?, last_room_version = GREATEST(last_room_version, ?), updated_at_ms = ?
WHERE app_code = ? AND mic_session_id = ? AND status <> ?
`, mictimedomain.SessionStatusEnded, endAt, endReason, seatOccupiedMs, micOnlineMs,
event.EventID, event.EventID, event.RoomVersion, event.OccurredAtMs, event.AppCode, event.MicSessionID, mictimedomain.SessionStatusEnded)
if err != nil {
return mictimedomain.ApplyResult{}, err
}
if err := incrementLifetime(ctx, tx, session, event, endAt, seatOccupiedMs, micOnlineMs); err != nil {
return mictimedomain.ApplyResult{}, err
}
if err := incrementDaily(ctx, tx, session, event, endAt); err != nil {
return mictimedomain.ApplyResult{}, err
}
return mictimedomain.ApplyResult{Consumed: true, Closed: true}, nil
}
func incrementLifetime(ctx context.Context, tx *sql.Tx, session micSessionRow, event mictimedomain.RoomMicEvent, endAt int64, seatMs int64, micMs int64) error {
_, err := tx.ExecContext(ctx, `
INSERT INTO user_mic_lifetime_stats (
app_code, user_id, seat_occupied_ms, mic_online_ms, session_count, first_mic_at_ms, last_mic_at_ms, updated_at_ms
) VALUES (?, ?, ?, ?, 1, ?, ?, ?)
ON DUPLICATE KEY UPDATE
seat_occupied_ms = seat_occupied_ms + VALUES(seat_occupied_ms),
mic_online_ms = mic_online_ms + VALUES(mic_online_ms),
session_count = session_count + 1,
first_mic_at_ms = CASE WHEN first_mic_at_ms = 0 OR first_mic_at_ms > VALUES(first_mic_at_ms) THEN VALUES(first_mic_at_ms) ELSE first_mic_at_ms END,
last_mic_at_ms = GREATEST(last_mic_at_ms, VALUES(last_mic_at_ms)),
updated_at_ms = VALUES(updated_at_ms)
`, event.AppCode, session.UserID, seatMs, micMs, session.SeatStartedAtMs, endAt, event.OccurredAtMs)
return err
}
func incrementDaily(ctx context.Context, tx *sql.Tx, session micSessionRow, event mictimedomain.RoomMicEvent, endAt int64) error {
seatParts := splitByUTCDay(session.SeatStartedAtMs, endAt)
micParts := map[string]dayDuration{}
if session.PublishingStartedAtMs.Valid {
publishStart := session.PublishingStartedAtMs.Int64
if publishStart < session.SeatStartedAtMs {
publishStart = session.SeatStartedAtMs
}
micParts = splitByUTCDay(publishStart, endAt)
}
dates := map[string]struct{}{}
for date := range seatParts {
dates[date] = struct{}{}
}
for date := range micParts {
dates[date] = struct{}{}
}
for date := range dates {
seatPart := seatParts[date]
micPart := micParts[date]
firstAt := minPositive(seatPart.FirstAtMs, micPart.FirstAtMs)
lastAt := maxInt64(seatPart.LastAtMs, micPart.LastAtMs)
if firstAt == 0 || lastAt == 0 {
continue
}
sessionCount := int64(0)
roomCount := int64(0)
if seatPart.DurationMs > 0 {
// A cross-day session is visible on every day it occupied a seat.
sessionCount = 1
roomCount = 1
}
_, err := tx.ExecContext(ctx, `
INSERT INTO user_mic_daily_stats (
app_code, user_id, stat_date, seat_occupied_ms, mic_online_ms, session_count, room_count,
first_mic_at_ms, last_mic_at_ms, updated_at_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
seat_occupied_ms = seat_occupied_ms + VALUES(seat_occupied_ms),
mic_online_ms = mic_online_ms + VALUES(mic_online_ms),
session_count = session_count + VALUES(session_count),
room_count = room_count + VALUES(room_count),
first_mic_at_ms = CASE WHEN first_mic_at_ms = 0 OR first_mic_at_ms > VALUES(first_mic_at_ms) THEN VALUES(first_mic_at_ms) ELSE first_mic_at_ms END,
last_mic_at_ms = GREATEST(last_mic_at_ms, VALUES(last_mic_at_ms)),
updated_at_ms = VALUES(updated_at_ms)
`, event.AppCode, session.UserID, date, seatPart.DurationMs, micPart.DurationMs, sessionCount, roomCount, firstAt, lastAt, event.OccurredAtMs)
if err != nil {
return err
}
}
return nil
}
type micSessionRow struct {
UserID int64
RoomID string
FirstSeatNo int32
CurrentSeatNo int32
Status string
SeatStartedAtMs int64
PublishingStartedAtMs sql.NullInt64
EndedAtMs sql.NullInt64
LastRoomVersion int64
}
func sessionForUpdate(ctx context.Context, tx *sql.Tx, appCode string, micSessionID string) (micSessionRow, bool, error) {
var session micSessionRow
err := tx.QueryRowContext(ctx, `
SELECT user_id, room_id, first_seat_no, current_seat_no, status, seat_started_at_ms,
publishing_started_at_ms, ended_at_ms, last_room_version
FROM user_mic_sessions
WHERE app_code = ? AND mic_session_id = ?
FOR UPDATE
`, appCode, micSessionID).Scan(&session.UserID, &session.RoomID, &session.FirstSeatNo, &session.CurrentSeatNo, &session.Status,
&session.SeatStartedAtMs, &session.PublishingStartedAtMs, &session.EndedAtMs, &session.LastRoomVersion)
if err == sql.ErrNoRows {
return micSessionRow{}, false, nil
}
if err != nil {
return micSessionRow{}, false, err
}
return session, true, nil
}
func updateSessionLastEvent(ctx context.Context, tx *sql.Tx, event mictimedomain.RoomMicEvent) error {
_, err := tx.ExecContext(ctx, `
UPDATE user_mic_sessions
SET last_event_id = ?, last_room_version = GREATEST(last_room_version, ?), updated_at_ms = ?
WHERE app_code = ? AND mic_session_id = ?
`, event.EventID, event.RoomVersion, event.OccurredAtMs, event.AppCode, event.MicSessionID)
return err
}
func insertEventConsumption(ctx context.Context, tx *sql.Tx, event mictimedomain.RoomMicEvent) error {
_, err := tx.ExecContext(ctx, `
INSERT INTO user_mic_event_consumption (
app_code, event_id, event_type, mic_session_id, user_id, action, status, skip_reason,
created_at_ms, consumed_at_ms, updated_at_ms
) VALUES (?, ?, ?, ?, ?, ?, 'processing', '', ?, NULL, ?)
`, event.AppCode, event.EventID, event.EventType, event.MicSessionID, event.TargetUserID, event.Action, event.OccurredAtMs, event.OccurredAtMs)
return err
}
func finishEventConsumption(ctx context.Context, tx *sql.Tx, event mictimedomain.RoomMicEvent, status string, reason string) error {
_, err := tx.ExecContext(ctx, `
UPDATE user_mic_event_consumption
SET status = ?, skip_reason = ?, consumed_at_ms = ?, updated_at_ms = ?
WHERE app_code = ? AND event_id = ?
`, status, strings.TrimSpace(reason), event.OccurredAtMs, event.OccurredAtMs, event.AppCode, event.EventID)
return err
}
func normalizeRoomMicEvent(event mictimedomain.RoomMicEvent) mictimedomain.RoomMicEvent {
event.AppCode = appcode.Normalize(event.AppCode)
event.EventID = strings.TrimSpace(event.EventID)
event.EventType = strings.TrimSpace(event.EventType)
event.RoomID = strings.TrimSpace(event.RoomID)
event.Action = strings.TrimSpace(event.Action)
event.MicSessionID = strings.TrimSpace(event.MicSessionID)
event.PublishState = strings.TrimSpace(event.PublishState)
event.Reason = strings.TrimSpace(event.Reason)
return event
}
type dayDuration struct {
DurationMs int64
FirstAtMs int64
LastAtMs int64
}
func splitByUTCDay(startMs int64, endMs int64) map[string]dayDuration {
parts := map[string]dayDuration{}
if endMs <= startMs {
return parts
}
for cursor := startMs; cursor < endMs; {
cursorTime := time.UnixMilli(cursor).UTC()
dayStart := time.Date(cursorTime.Year(), cursorTime.Month(), cursorTime.Day(), 0, 0, 0, 0, time.UTC)
nextDayMs := dayStart.AddDate(0, 0, 1).UnixMilli()
segmentEnd := minInt64(endMs, nextDayMs)
date := dayStart.Format("2006-01-02")
part := parts[date]
if part.FirstAtMs == 0 || cursor < part.FirstAtMs {
part.FirstAtMs = cursor
}
if segmentEnd > part.LastAtMs {
part.LastAtMs = segmentEnd
}
part.DurationMs += segmentEnd - cursor
parts[date] = part
cursor = segmentEnd
}
return parts
}
func isDuplicate(err error) bool {
var mysqlErr *mysqldriver.MySQLError
return errors.As(err, &mysqlErr) && mysqlErr.Number == 1062
}
func minPositive(a int64, b int64) int64 {
if a <= 0 {
return b
}
if b <= 0 || a < b {
return a
}
return b
}
func minInt64(a int64, b int64) int64 {
if a < b {
return a
}
return b
}
func maxInt64(a int64, b int64) int64 {
if a > b {
return a
}
return b
}