feat: complete IM P0 and P1 reliability work

2026-08-17 13:43:27 +08:00
parent ddf0f7a55e
commit f2c229135b
43 changed files with 4512 additions and 1181 deletions
@@ -0,0 +1,253 @@
+//go:build mysql
+// +build mysql
+
+package mysql
+
+import (
+	"database/sql"
+	"encoding/json"
+	"time"
+
+	"github.com/jmoiron/sqlx"
+	"github.com/tinode/chat/server/store"
+	t "github.com/tinode/chat/server/store/types"
+)
+
+func (a *adapter) infrastructureAdapter(id t.TenantID) (*adapter, error) {
+	if !a.tenantScopedDB {
+		return a.routedTenantAdapter(id)
+	}
+	if err := validTenant(id); err != nil {
+		return nil, err
+	}
+	return a, nil
+}
+
+func (a *adapter) TenantHTTPIdempotencyBegin(id t.TenantID, record *t.HTTPIdempotencyRecord) (*t.HTTPIdempotencyRecord, bool, error) {
+	tenantAdp, err := a.infrastructureAdapter(id)
+	if err != nil {
+		return nil, false, err
+	}
+	if record == nil || record.UserID.IsZero() || record.Method == "" || record.Route == "" || record.Key == "" || record.BodyHash == "" {
+		return nil, false, t.ErrMalformed
+	}
+	if tenantAdp != a {
+		return tenantAdp.TenantHTTPIdempotencyBegin(id, record)
+	}
+	now := t.TimeNow()
+	if record.ExpiresAt.IsZero() {
+		record.ExpiresAt = now.Add(24 * time.Hour)
+	}
+	_, err = a.db.Exec(`INSERT INTO im_http_idempotency
+		(user_id,method,route,idempotency_key,body_hash,state,response_status,response_body,expires_at,created_at,updated_at)
+		VALUES(?,?,?,?,?,?,?,?,?,?,?)`, decodeUID(record.UserID), record.Method, record.Route, record.Key,
+		record.BodyHash, t.HTTPIdempotencyProcessing, 0, nil, record.ExpiresAt, now, now)
+	if err == nil {
+		record.State = t.HTTPIdempotencyProcessing
+		record.CreatedAt = now
+		record.UpdatedAt = now
+		return record, true, nil
+	}
+	if !isDupe(err) {
+		return nil, false, err
+	}
+	existing, err := a.httpIdempotencyGet(record.UserID, record.Method, record.Route, record.Key)
+	if err == nil && !existing.ExpiresAt.After(now) {
+		res, deleteErr := a.db.Exec(`DELETE FROM im_http_idempotency
+			WHERE user_id=? AND method=? AND route=? AND idempotency_key=? AND expires_at<=?`,
+			decodeUID(record.UserID), record.Method, record.Route, record.Key, now)
+		if deleteErr != nil {
+			return nil, false, deleteErr
+		}
+		if deleted, _ := res.RowsAffected(); deleted > 0 {
+			return a.TenantHTTPIdempotencyBegin(id, record)
+		}
+		current, getErr := a.httpIdempotencyGet(record.UserID, record.Method, record.Route, record.Key)
+		return current, false, getErr
+	}
+	return existing, false, err
+}
+
+func (a *adapter) TenantHTTPIdempotencyComplete(id t.TenantID, record *t.HTTPIdempotencyRecord) error {
+	tenantAdp, err := a.infrastructureAdapter(id)
+	if err != nil {
+		return err
+	}
+	if record == nil || record.UserID.IsZero() {
+		return t.ErrMalformed
+	}
+	if tenantAdp != a {
+		return tenantAdp.TenantHTTPIdempotencyComplete(id, record)
+	}
+	now := t.TimeNow()
+	res, err := a.db.Exec(`UPDATE im_http_idempotency SET state=?,response_status=?,response_body=?,updated_at=?
+		WHERE user_id=? AND method=? AND route=? AND idempotency_key=? AND body_hash=? AND state=?`,
+		t.HTTPIdempotencyCompleted, record.ResponseStatus, record.ResponseBody, now, decodeUID(record.UserID),
+		record.Method, record.Route, record.Key, record.BodyHash, t.HTTPIdempotencyProcessing)
+	if err != nil {
+		return err
+	}
+	if count, _ := res.RowsAffected(); count == 0 {
+		return t.ErrNotFound
+	}
+	return nil
+}
+
+func (a *adapter) TenantHTTPIdempotencyAbort(id t.TenantID, record *t.HTTPIdempotencyRecord) error {
+	tenantAdp, err := a.infrastructureAdapter(id)
+	if err != nil {
+		return err
+	}
+	if record == nil || record.UserID.IsZero() {
+		return t.ErrMalformed
+	}
+	if tenantAdp != a {
+		return tenantAdp.TenantHTTPIdempotencyAbort(id, record)
+	}
+	_, err = a.db.Exec(`DELETE FROM im_http_idempotency
+		WHERE user_id=? AND method=? AND route=? AND idempotency_key=? AND body_hash=? AND state=?`,
+		decodeUID(record.UserID), record.Method, record.Route, record.Key, record.BodyHash, t.HTTPIdempotencyProcessing)
+	return err
+}
+
+func (a *adapter) httpIdempotencyGet(uid t.Uid, method, route, key string) (*t.HTTPIdempotencyRecord, error) {
+	row := struct {
+		UserID         int64 `db:"user_id"`
+		Method         string
+		Route          string
+		Key            string `db:"idempotency_key"`
+		BodyHash       string `db:"body_hash"`
+		State          string
+		ResponseStatus int       `db:"response_status"`
+		ResponseBody   []byte    `db:"response_body"`
+		CreatedAt      time.Time `db:"created_at"`
+		UpdatedAt      time.Time `db:"updated_at"`
+		ExpiresAt      time.Time `db:"expires_at"`
+	}{}
+	err := a.db.Get(&row, `SELECT user_id,method,route,idempotency_key,body_hash,state,response_status,response_body,created_at,updated_at,expires_at
+		FROM im_http_idempotency WHERE user_id=? AND method=? AND route=? AND idempotency_key=?`, decodeUID(uid), method, route, key)
+	if err == sql.ErrNoRows {
+		return nil, t.ErrNotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &t.HTTPIdempotencyRecord{UserID: store.EncodeUid(row.UserID), Method: row.Method, Route: row.Route,
+		Key: row.Key, BodyHash: row.BodyHash, State: row.State, ResponseStatus: row.ResponseStatus,
+		ResponseBody: row.ResponseBody, CreatedAt: row.CreatedAt, UpdatedAt: row.UpdatedAt, ExpiresAt: row.ExpiresAt}, nil
+}
+
+func (a *adapter) TenantBusinessEventCreate(id t.TenantID, event *t.BusinessEvent) error {
+	tenantAdp, err := a.infrastructureAdapter(id)
+	if err != nil {
+		return err
+	}
+	if event == nil || event.EventID == "" || event.Target == "" || event.EventType == "" ||
+		(event.TargetType != "user" && event.TargetType != "topic") {
+		return t.ErrMalformed
+	}
+	if tenantAdp != a {
+		return tenantAdp.TenantBusinessEventCreate(id, event)
+	}
+	now := t.TimeNow()
+	_, err = a.db.Exec(`INSERT INTO im_business_events_outbox
+		(event_id,target_type,target,actor_user_id,event_type,payload,status,created_at,updated_at)
+		VALUES(?,?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, event.EventID,
+		event.TargetType, event.Target, decodeUID(event.ActorUserID), event.EventType, event.Payload,
+		t.BusinessEventPending, now, now)
+	return err
+}
+
+func (a *adapter) TenantBusinessEventListPending(id t.TenantID, limit int, now time.Time) ([]t.BusinessEvent, error) {
+	tenantAdp, err := a.infrastructureAdapter(id)
+	if err != nil {
+		return nil, err
+	}
+	if tenantAdp != a {
+		return tenantAdp.TenantBusinessEventListPending(id, limit, now)
+	}
+	if limit <= 0 || limit > 500 {
+		limit = 100
+	}
+	rows, err := a.db.Queryx(`SELECT * FROM im_business_events_outbox
+		WHERE status=? AND (next_retry_at IS NULL OR next_retry_at<=?) ORDER BY id LIMIT ?`, t.BusinessEventPending, now, limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	events := make([]t.BusinessEvent, 0, limit)
+	for rows.Next() {
+		var row struct {
+			ID          int64
+			EventID     string `db:"event_id"`
+			TargetType  string `db:"target_type"`
+			Target      string
+			ActorUserID int64  `db:"actor_user_id"`
+			EventType   string `db:"event_type"`
+			Payload     []byte
+			Status      string
+			Attempts    int
+			NextRetryAt *time.Time `db:"next_retry_at"`
+			DeliveredAt *time.Time `db:"delivered_at"`
+			CreatedAt   time.Time  `db:"created_at"`
+			UpdatedAt   time.Time  `db:"updated_at"`
+		}
+		if err = rows.StructScan(&row); err != nil {
+			return nil, err
+		}
+		events = append(events, t.BusinessEvent{ID: row.ID, EventID: row.EventID, TargetType: row.TargetType,
+			Target: row.Target, ActorUserID: store.EncodeUid(row.ActorUserID), EventType: row.EventType,
+			Payload: row.Payload, Status: row.Status, Attempts: row.Attempts, NextRetryAt: row.NextRetryAt,
+			DeliveredAt: row.DeliveredAt, CreatedAt: row.CreatedAt, UpdatedAt: row.UpdatedAt})
+	}
+	return events, rows.Err()
+}
+
+func (a *adapter) TenantBusinessEventMarkDelivered(id t.TenantID, eventID string, now time.Time) error {
+	tenantAdp, err := a.infrastructureAdapter(id)
+	if err != nil {
+		return err
+	}
+	if eventID == "" {
+		return t.ErrMalformed
+	}
+	if tenantAdp != a {
+		return tenantAdp.TenantBusinessEventMarkDelivered(id, eventID, now)
+	}
+	res, err := a.db.Exec(`UPDATE im_business_events_outbox SET status=?,delivered_at=?,attempts=attempts+1,updated_at=?
+		WHERE event_id=? AND status=?`, t.BusinessEventDelivered, now, now, eventID, t.BusinessEventPending)
+	if err != nil {
+		return err
+	}
+	if count, _ := res.RowsAffected(); count == 0 {
+		return t.ErrNotFound
+	}
+	return nil
+}
+
+func insertBusinessEventTx(tx *sqlx.Tx, eventID, targetType, target string, actor t.Uid, eventType string, payload any) error {
+	encoded, err := json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+	now := t.TimeNow()
+	_, err = tx.Exec(`INSERT INTO im_business_events_outbox
+		(event_id,target_type,target,actor_user_id,event_type,payload,status,created_at,updated_at)
+		VALUES(?,?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE event_id=VALUES(event_id)`, eventID, targetType,
+		target, decodeUID(actor), eventType, encoded, t.BusinessEventPending, now, now)
+	return err
+}
+
+func insertSystemNotificationTx(tx *sqlx.Tx, uid t.Uid, eventID, eventType, title, summary string, action any) error {
+	encoded, err := json.Marshal(action)
+	if err != nil {
+		return err
+	}
+	now := t.TimeNow()
+	_, err = tx.Exec(`INSERT INTO im_system_notifications
+		(user_id,event_id,event_type,title,summary,action,created_at,updated_at)
+		VALUES(?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE event_type=VALUES(event_type),title=VALUES(title),
+		summary=VALUES(summary),action=VALUES(action),updated_at=VALUES(updated_at)`, decodeUID(uid), eventID,
+		eventType, title, summary, encoded, now, now)
+	return err
+}