im_server/server/db/mysql/tenant_infrastructure.go

//go:build mysql
// +build mysql

package mysql

import (
	"encoding/json"
	"strconv"
	"strings"
	"time"

	"github.com/jmoiron/sqlx"
	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) TenantKafkaConsumeDead(id t.TenantID, dead *t.KafkaDeadLetter, now time.Time) error {
	tenantAdp, err := a.infrastructureAdapter(id)
	if err != nil {
		return err
	}
	if dead == nil || dead.DeadID == "" || dead.EventID == "" || dead.ConsumerGroup == "" || dead.SourceTopic == "" || len(dead.Envelope) == 0 {
		return t.ErrMalformed
	}
	if tenantAdp != a {
		return tenantAdp.TenantKafkaConsumeDead(id, dead, now)
	}
	if _, err = a.db.Exec(`INSERT INTO im_kafka_dead_letters
		(dead_id,event_id,parent_event_id,source_topic,consumer_group,event_type,biz_type,biz_id,envelope,
		 retry_count,last_reason,last_error,status,replay_count,created_at,updated_at)
		VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
		ON DUPLICATE KEY UPDATE last_reason=VALUES(last_reason),last_error=VALUES(last_error),updated_at=VALUES(updated_at)`,
		dead.DeadID, dead.EventID, dead.ParentEventID, dead.SourceTopic, dead.ConsumerGroup, dead.EventType,
		dead.BizType, dead.BizID, dead.Envelope, dead.RetryCount, truncateKafkaString(dead.LastReason, 64),
		truncateKafkaString(dead.LastError, 1000), t.KafkaDeadStatusDead, dead.ReplayCount, now, now); err != nil {
		return err
	}
	return nil
}

func (a *adapter) TenantKafkaDeadLetterList(id t.TenantID, opts t.KafkaDeadLetterListOptions) (*t.KafkaDeadLetterPage, error) {
	tenantAdp, err := a.infrastructureAdapter(id)
	if err != nil {
		return nil, err
	}
	if tenantAdp != a {
		return tenantAdp.TenantKafkaDeadLetterList(id, opts)
	}
	limit := opts.Limit
	if limit <= 0 || limit > 200 {
		limit = 50
	}
	where := []string{"1=1"}
	args := []any{}
	if opts.Status != "" {
		where = append(where, "status=?")
		args = append(args, opts.Status)
	}
	if opts.BizType != "" {
		where = append(where, "biz_type=?")
		args = append(args, opts.BizType)
	}
	if opts.BizID != "" {
		where = append(where, "biz_id=?")
		args = append(args, opts.BizID)
	}
	if opts.EventType != "" {
		where = append(where, "event_type=?")
		args = append(args, opts.EventType)
	}
	if opts.Cursor != "" {
		where = append(where, "id<?")
		args = append(args, opts.Cursor)
	}
	args = append(args, limit+1)
	rows, err := a.db.Queryx(`SELECT id,dead_id,event_id,parent_event_id,source_topic,consumer_group,event_type,biz_type,biz_id,
		envelope,retry_count,last_reason,last_error,status,replay_count,created_at,updated_at
		FROM im_kafka_dead_letters WHERE `+strings.Join(where, " AND ")+` ORDER BY id DESC LIMIT ?`, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	page := &t.KafkaDeadLetterPage{Items: make([]t.KafkaDeadLetter, 0, limit)}
	for rows.Next() {
		dead, err := scanKafkaDeadLetter(rows)
		if err != nil {
			return nil, err
		}
		if len(page.Items) == limit {
			page.NextCursor = strconv.FormatInt(dead.ID, 10)
			continue
		}
		page.Items = append(page.Items, *dead)
	}
	return page, rows.Err()
}

func (a *adapter) TenantKafkaDeadLetterGet(id t.TenantID, deadID string) (*t.KafkaDeadLetter, error) {
	tenantAdp, err := a.infrastructureAdapter(id)
	if err != nil {
		return nil, err
	}
	if strings.TrimSpace(deadID) == "" {
		return nil, t.ErrMalformed
	}
	if tenantAdp != a {
		return tenantAdp.TenantKafkaDeadLetterGet(id, deadID)
	}
	rows, err := a.db.Queryx(`SELECT id,dead_id,event_id,parent_event_id,source_topic,consumer_group,event_type,biz_type,biz_id,
		envelope,retry_count,last_reason,last_error,status,replay_count,created_at,updated_at
		FROM im_kafka_dead_letters WHERE dead_id=?`, strings.TrimSpace(deadID))
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	if rows.Next() {
		return scanKafkaDeadLetter(rows)
	}
	if err = rows.Err(); err != nil {
		return nil, err
	}
	return nil, t.ErrNotFound
}

func (a *adapter) TenantKafkaDeadLetterMark(id t.TenantID, deadID, status, resolvedBy, note string, now time.Time) error {
	tenantAdp, err := a.infrastructureAdapter(id)
	if err != nil {
		return err
	}
	deadID = strings.TrimSpace(deadID)
	status = strings.TrimSpace(status)
	if deadID == "" || (status != t.KafkaDeadStatusIgnored && status != t.KafkaDeadStatusResolved && status != t.KafkaDeadStatusReplayed) {
		return t.ErrMalformed
	}
	if tenantAdp != a {
		return tenantAdp.TenantKafkaDeadLetterMark(id, deadID, status, resolvedBy, note, now)
	}
	res, err := a.db.Exec(`UPDATE im_kafka_dead_letters SET status=?,resolved_by=?,resolved_at=?,resolve_note=?,updated_at=?
		WHERE dead_id=? AND status IN (?,?,?)`, status, truncateKafkaString(resolvedBy, 64), now,
		truncateKafkaString(note, 1000), now, deadID, t.KafkaDeadStatusDead, t.KafkaDeadStatusPendingReplay, t.KafkaDeadStatusReplayed)
	if err != nil {
		return err
	}
	if affected, _ := res.RowsAffected(); affected == 0 {
		return t.ErrNotFound
	}
	return nil
}

func (a *adapter) TenantKafkaDeadLetterMarkReplay(id t.TenantID, deadID, resolvedBy, note string, now time.Time) (*t.KafkaDeadLetter, error) {
	tenantAdp, err := a.infrastructureAdapter(id)
	if err != nil {
		return nil, err
	}
	deadID = strings.TrimSpace(deadID)
	if deadID == "" {
		return nil, t.ErrMalformed
	}
	if tenantAdp != a {
		return tenantAdp.TenantKafkaDeadLetterMarkReplay(id, deadID, resolvedBy, note, now)
	}
	res, err := a.db.Exec(`UPDATE im_kafka_dead_letters SET status=?,replay_count=replay_count+1,resolved_by=?,
		resolved_at=?,resolve_note=?,updated_at=? WHERE dead_id=? AND status IN (?,?)`,
		t.KafkaDeadStatusPendingReplay, truncateKafkaString(resolvedBy, 64), now, truncateKafkaString(note, 1000),
		now, deadID, t.KafkaDeadStatusDead, t.KafkaDeadStatusPendingReplay)
	if err != nil {
		return nil, err
	}
	if affected, _ := res.RowsAffected(); affected == 0 {
		return nil, t.ErrNotFound
	}
	return a.TenantKafkaDeadLetterGet(id, deadID)
}

func scanKafkaDeadLetter(rows *sqlx.Rows) (*t.KafkaDeadLetter, error) {
	var row struct {
		ID            int64     `db:"id"`
		DeadID        string    `db:"dead_id"`
		EventID       string    `db:"event_id"`
		ParentEventID string    `db:"parent_event_id"`
		SourceTopic   string    `db:"source_topic"`
		ConsumerGroup string    `db:"consumer_group"`
		EventType     string    `db:"event_type"`
		BizType       string    `db:"biz_type"`
		BizID         string    `db:"biz_id"`
		Envelope      []byte    `db:"envelope"`
		RetryCount    int       `db:"retry_count"`
		LastReason    string    `db:"last_reason"`
		LastError     string    `db:"last_error"`
		Status        string    `db:"status"`
		ReplayCount   int       `db:"replay_count"`
		CreatedAt     time.Time `db:"created_at"`
		UpdatedAt     time.Time `db:"updated_at"`
	}
	if err := rows.StructScan(&row); err != nil {
		return nil, err
	}
	return &t.KafkaDeadLetter{
		ID:            row.ID,
		DeadID:        row.DeadID,
		EventID:       row.EventID,
		ParentEventID: row.ParentEventID,
		SourceTopic:   row.SourceTopic,
		ConsumerGroup: row.ConsumerGroup,
		EventType:     row.EventType,
		BizType:       row.BizType,
		BizID:         row.BizID,
		Envelope:      row.Envelope,
		RetryCount:    row.RetryCount,
		LastReason:    row.LastReason,
		LastError:     row.LastError,
		Status:        row.Status,
		ReplayCount:   row.ReplayCount,
		CreatedAt:     row.CreatedAt,
		UpdatedAt:     row.UpdatedAt,
	}, nil
}

func truncateKafkaString(value string, limit int) string {
	if limit <= 0 {
		return ""
	}
	runes := []rune(value)
	if len(runes) <= limit {
		return value
	}
	return string(runes[:limit])
}

func insertSystemNotificationTx(tx *sqlx.Tx, tenantID t.TenantID, 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(tenantShardSQL(tenantID, `INSERT INTO im_system_notifications
		(tenant_id,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)`), tenantID, decodeUID(uid), eventID,
		eventType, title, summary, encoded, now, now)
	return err
}