im_server/server/db/mysql/tenant_infrastructure.go

//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
}