NixMsg/internal/app/group/void.go

package group

import (
	"context"
	"database/sql"
	"strings"

	"git.asio.asia/nixevol/NixMsg/internal/app/port"
	"git.asio.asia/nixevol/NixMsg/internal/protocol"
)

type revokeItem struct {
	endpointID string
	msgID      string
	fromID     string
	reason     string
}

// voidMemberDeliveriesTx rejects pending deliveries for a leaving member; records revokes for pushed ones.
func voidMemberDeliveriesTx(tx *sql.Tx, groupID, endpointID, reason string, nowMs int64, revokes *[]revokeItem) error {
	rows, err := tx.Query(`
SELECT d.seq, d.pushed_at, m.id, m.sender_id
FROM deliveries d
JOIN messages m ON m.seq = d.seq
WHERE d.endpoint_id = ? AND d.state = 'pending'
  AND m.dest_kind = 'group' AND m.dest_id = ?`, endpointID, groupID)
	if err != nil {
		return err
	}
	defer func() { _ = rows.Close() }()
	type row struct {
		seq      int64
		pushed   sql.NullInt64
		msgID    string
		senderID string
	}
	var list []row
	for rows.Next() {
		var r row
		if scanErr := rows.Scan(&r.seq, &r.pushed, &r.msgID, &r.senderID); scanErr != nil {
			return scanErr
		}
		list = append(list, r)
	}
	if err = rows.Err(); err != nil {
		return err
	}
	for _, r := range list {
		if _, execErr := tx.Exec(`
UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`,
			reason, nowMs, r.seq, endpointID); execErr != nil {
			return execErr
		}
		if r.pushed.Valid && revokes != nil {
			*revokes = append(*revokes, revokeItem{
				endpointID: endpointID, msgID: r.msgID, fromID: r.senderID, reason: reason,
			})
		}
	}
	return nil
}

// voidGroupAllTx rejects all pending group deliveries and completes scheduled messages.
func voidGroupAllTx(tx *sql.Tx, groupID string, nowMs int64, revokes *[]revokeItem) error {
	rows, err := tx.Query(`
SELECT d.seq, d.endpoint_id, d.pushed_at, m.id, m.sender_id
FROM deliveries d
JOIN messages m ON m.seq = d.seq
WHERE d.state = 'pending' AND m.dest_kind = 'group' AND m.dest_id = ?`, groupID)
	if err != nil {
		return err
	}
	type drow struct {
		seq        int64
		endpointID string
		pushed     sql.NullInt64
		msgID      string
		senderID   string
	}
	var dlist []drow
	for rows.Next() {
		var r drow
		if scanErr := rows.Scan(&r.seq, &r.endpointID, &r.pushed, &r.msgID, &r.senderID); scanErr != nil {
			_ = rows.Close()
			return scanErr
		}
		dlist = append(dlist, r)
	}
	_ = rows.Close()
	if err = rows.Err(); err != nil {
		return err
	}
	for _, r := range dlist {
		if _, execErr := tx.Exec(`
UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ?
WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`,
			reasonGroupDissolved, nowMs, r.seq, r.endpointID); execErr != nil {
			return execErr
		}
		if r.pushed.Valid && revokes != nil {
			*revokes = append(*revokes, revokeItem{
				endpointID: r.endpointID, msgID: r.msgID, fromID: r.senderID, reason: reasonGroupDissolved,
			})
		}
	}

	srows, err := tx.Query(`
SELECT seq, id, sender_id, receipt FROM messages
WHERE dest_kind = 'group' AND dest_id = ? AND state = 'scheduled'`, groupID)
	if err != nil {
		return err
	}
	type srow struct {
		seq      int64
		msgID    string
		senderID string
		receipt  int
	}
	var slist []srow
	for srows.Next() {
		var r srow
		if scanErr := srows.Scan(&r.seq, &r.msgID, &r.senderID, &r.receipt); scanErr != nil {
			_ = srows.Close()
			return scanErr
		}
		slist = append(slist, r)
	}
	_ = srows.Close()
	if err = srows.Err(); err != nil {
		return err
	}
	for _, r := range slist {
		if _, execErr := tx.Exec(`
UPDATE messages SET state = 'completed', reason = ? WHERE seq = ? AND state = 'scheduled'`,
			reasonGroupDissolved, r.seq); execErr != nil {
			return execErr
		}
		if _, execErr := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, r.seq); execErr != nil {
			return execErr
		}
		if r.receipt != 0 {
			if _, execErr := tx.Exec(`
INSERT INTO receipts(sender_id, msg_id, endpoint_id, state, reason, created_at, acked)
VALUES(?,?,?,?,?,?,0)`,
				r.senderID, r.msgID, "", "completed", reasonGroupDissolved, nowMs); execErr != nil {
				return execErr
			}
		}
	}
	return nil
}

func (a *App) publishRevokes(ctx context.Context, items []revokeItem) {
	if a.down == nil || len(items) == 0 {
		return
	}
	for _, it := range items {
		frame := protocol.Revoked{
			V: protocol.Version, Type: protocol.TypeRevoked,
			ID: it.msgID, From: it.fromID, Reason: it.reason,
		}
		payload, encErr := encodeFrame(frame)
		if encErr != nil {
			continue
		}
		_ = a.down.PublishDown(ctx, it.endpointID, "", payload, port.PublishOpts{QoS: 1})
	}
}

func isUnique(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "unique") || strings.Contains(msg, "constraint failed")
}