goclaw/internal/store/pg/pairing.go

package pg

import (
	"context"
	"crypto/rand"
	"database/sql"
	"encoding/json"
	"fmt"
	"time"

	"github.com/google/uuid"

	"github.com/nextlevelbuilder/goclaw/internal/store"
)

const (
	codeAlphabet         = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"
	codeLength           = 8
	codeTTL              = 60 * time.Minute
	pairedDeviceTTL      = 30 * 24 * time.Hour // 30 days
	maxPendingPerAccount = 3
)

// PGPairingStore implements store.PairingStore backed by Postgres.
type PGPairingStore struct {
	db        *sql.DB
	onRequest func(code, senderID, channel, chatID string)
}

func NewPGPairingStore(db *sql.DB) *PGPairingStore {
	return &PGPairingStore{db: db}
}

// SetOnRequest sets a callback fired after a new pairing request is created.
func (s *PGPairingStore) SetOnRequest(cb func(code, senderID, channel, chatID string)) {
	s.onRequest = cb
}

func (s *PGPairingStore) RequestPairing(ctx context.Context, senderID, channel, chatID, accountID string, metadata map[string]string) (string, error) {
	tid := tenantIDForInsert(ctx)

	// Prune expired
	s.db.ExecContext(ctx, "DELETE FROM pairing_requests WHERE expires_at < $1", time.Now())

	// Check max pending (per tenant)
	var count int64
	s.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM pairing_requests WHERE account_id = $1 AND tenant_id = $2", accountID, tid).Scan(&count)
	if count >= maxPendingPerAccount {
		return "", fmt.Errorf("max pending pairing requests (%d) exceeded", maxPendingPerAccount)
	}

	// Check existing (per tenant)
	var existingCode string
	err := s.db.QueryRowContext(ctx, "SELECT code FROM pairing_requests WHERE sender_id = $1 AND channel = $2 AND tenant_id = $3", senderID, channel, tid).Scan(&existingCode)
	if err == nil {
		return existingCode, nil
	}

	metaJSON := []byte("{}")
	if len(metadata) > 0 {
		metaJSON, _ = json.Marshal(metadata)
	}

	code := generatePairingCode()
	now := time.Now()
	_, err = s.db.ExecContext(ctx,
		`INSERT INTO pairing_requests (id, code, sender_id, channel, chat_id, account_id, expires_at, created_at, metadata, tenant_id)
		 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`,
		uuid.Must(uuid.NewV7()), code, senderID, channel, chatID, accountID, now.Add(codeTTL), now, metaJSON, tid,
	)
	if err != nil {
		return "", fmt.Errorf("create pairing request: %w", err)
	}
	if s.onRequest != nil {
		go s.onRequest(code, senderID, channel, chatID)
	}
	return code, nil
}

// ApprovePairing looks up by code (globally unique random token) and creates paired device.
// The approver's tenant context determines paired_devices.tenant_id.
func (s *PGPairingStore) ApprovePairing(ctx context.Context, code, approvedBy string) (*store.PairedDeviceData, error) {
	// Prune expired
	s.db.ExecContext(ctx, "DELETE FROM pairing_requests WHERE expires_at < $1", time.Now())

	var reqID uuid.UUID
	var senderID, channel, chatID string
	var metaJSON []byte
	var reqTenantID uuid.UUID
	// Code lookup is cross-tenant (random token, approver is WS/HTTP user).
	// Also check expires_at to close race between prune DELETE and this SELECT.
	err := s.db.QueryRowContext(ctx,
		"SELECT id, sender_id, channel, chat_id, COALESCE(metadata, '{}'), tenant_id FROM pairing_requests WHERE code = $1 AND expires_at > NOW()", code,
	).Scan(&reqID, &senderID, &channel, &chatID, &metaJSON, &reqTenantID)
	if err != nil {
		return nil, fmt.Errorf("pairing code %s not found or expired", code)
	}

	// Remove from pending
	s.db.ExecContext(ctx, "DELETE FROM pairing_requests WHERE id = $1", reqID)

	// Add to paired — use the request's tenant (the channel that initiated pairing)
	now := time.Now()
	expiresAt := now.Add(pairedDeviceTTL)
	_, err = s.db.ExecContext(ctx,
		`INSERT INTO paired_devices (id, sender_id, channel, chat_id, paired_by, paired_at, metadata, expires_at, tenant_id)
		 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)`,
		uuid.Must(uuid.NewV7()), senderID, channel, chatID, approvedBy, now, metaJSON, expiresAt, reqTenantID,
	)
	if err != nil {
		return nil, fmt.Errorf("create paired device: %w", err)
	}

	var meta map[string]string
	if len(metaJSON) > 0 {
		json.Unmarshal(metaJSON, &meta)
	}

	expiresAtMs := expiresAt.UnixMilli()
	return &store.PairedDeviceData{
		SenderID:  senderID,
		Channel:   channel,
		ChatID:    chatID,
		PairedAt:  now.UnixMilli(),
		PairedBy:  approvedBy,
		ExpiresAt: &expiresAtMs,
		Metadata:  meta,
	}, nil
}

func (s *PGPairingStore) DenyPairing(ctx context.Context, code string) error {
	// Code lookup is cross-tenant (random token)
	result, err := s.db.ExecContext(ctx, "DELETE FROM pairing_requests WHERE code = $1", code)
	if err != nil {
		return err
	}
	n, _ := result.RowsAffected()
	if n == 0 {
		return fmt.Errorf("pairing code %s not found or expired", code)
	}
	return nil
}

func (s *PGPairingStore) RevokePairing(ctx context.Context, senderID, channel string) error {
	tid := tenantIDForInsert(ctx)
	result, err := s.db.ExecContext(ctx, "DELETE FROM paired_devices WHERE sender_id = $1 AND channel = $2 AND tenant_id = $3", senderID, channel, tid)
	if err != nil {
		return err
	}
	n, _ := result.RowsAffected()
	if n == 0 {
		return fmt.Errorf("paired device not found: %s/%s", channel, senderID)
	}
	return nil
}

// SetPairingPermanent clears (permanent=true) or restarts (permanent=false)
// the expiry of a live pairing. An already expired pairing is not revived.
func (s *PGPairingStore) SetPairingPermanent(ctx context.Context, senderID, channel string, permanent bool) error {
	tid := tenantIDForInsert(ctx)
	var expiresAt *time.Time
	if !permanent {
		t := time.Now().Add(pairedDeviceTTL)
		expiresAt = &t
	}
	result, err := s.db.ExecContext(ctx,
		`UPDATE paired_devices SET expires_at = $1
		 WHERE sender_id = $2 AND channel = $3 AND tenant_id = $4 AND (expires_at IS NULL OR expires_at > NOW())`,
		expiresAt, senderID, channel, tid,
	)
	if err != nil {
		return err
	}
	n, _ := result.RowsAffected()
	if n == 0 {
		return fmt.Errorf("%w: %s/%s", store.ErrPairedDeviceNotFound, channel, senderID)
	}
	return nil
}

func (s *PGPairingStore) IsPaired(ctx context.Context, senderID, channel string) (bool, error) {
	tid := tenantIDForInsert(ctx)
	var count int64
	err := s.db.QueryRowContext(ctx,
		"SELECT COUNT(*) FROM paired_devices WHERE sender_id = $1 AND channel = $2 AND tenant_id = $3 AND (expires_at IS NULL OR expires_at > NOW())",
		senderID, channel, tid,
	).Scan(&count)
	if err != nil {
		return false, fmt.Errorf("pairing check query: %w", err)
	}
	return count > 0, nil
}

// pairingRequestRow is an sqlx scan struct for pairing_requests.
// Domain struct uses int64 (Unix ms) for timestamps, DB stores time.Time.
type pairingRequestRow struct {
	Code      string    `json:"code" db:"code"`
	SenderID  string    `json:"sender_id" db:"sender_id"`
	Channel   string    `json:"channel" db:"channel"`
	ChatID    string    `json:"chat_id" db:"chat_id"`
	AccountID string    `json:"account_id" db:"account_id"`
	CreatedAt time.Time `json:"created_at" db:"created_at"`
	ExpiresAt time.Time `json:"expires_at" db:"expires_at"`
	Metadata  []byte    `json:"metadata" db:"metadata"`
}

// pairedDeviceRow is an sqlx scan struct for paired_devices.
type pairedDeviceRow struct {
	SenderID  string     `json:"sender_id" db:"sender_id"`
	Channel   string     `json:"channel" db:"channel"`
	ChatID    string     `json:"chat_id" db:"chat_id"`
	PairedBy  string     `json:"paired_by" db:"paired_by"`
	PairedAt  time.Time  `json:"paired_at" db:"paired_at"`
	ExpiresAt *time.Time `json:"expires_at" db:"expires_at"`
	Metadata  []byte     `json:"metadata" db:"metadata"`
}

func (s *PGPairingStore) ListPending(ctx context.Context) []store.PairingRequestData {
	tid := tenantIDForInsert(ctx)

	// Prune expired
	s.db.ExecContext(ctx, "DELETE FROM pairing_requests WHERE expires_at < $1", time.Now())

	var rows []pairingRequestRow
	err := pkgSqlxDB.SelectContext(ctx, &rows,
		`SELECT code, sender_id, channel, chat_id, account_id, created_at, expires_at, COALESCE(metadata, '{}') AS metadata
		 FROM pairing_requests WHERE tenant_id = $1 ORDER BY created_at DESC`, tid)
	if err != nil {
		return []store.PairingRequestData{}
	}

	result := make([]store.PairingRequestData, len(rows))
	for i, r := range rows {
		result[i] = store.PairingRequestData{
			Code: r.Code, SenderID: r.SenderID, Channel: r.Channel,
			ChatID: r.ChatID, AccountID: r.AccountID,
			CreatedAt: r.CreatedAt.UnixMilli(), ExpiresAt: r.ExpiresAt.UnixMilli(),
		}
		if len(r.Metadata) > 0 {
			json.Unmarshal(r.Metadata, &result[i].Metadata)
		}
	}
	return result
}

func (s *PGPairingStore) ListPaired(ctx context.Context) []store.PairedDeviceData {
	tid := tenantIDForInsert(ctx)

	// Prune expired paired devices
	s.db.ExecContext(ctx, "DELETE FROM paired_devices WHERE expires_at IS NOT NULL AND expires_at < NOW()")

	var rows []pairedDeviceRow
	err := pkgSqlxDB.SelectContext(ctx, &rows,
		`SELECT sender_id, channel, chat_id, paired_by, paired_at, expires_at, COALESCE(metadata, '{}') AS metadata
		 FROM paired_devices WHERE tenant_id = $1 ORDER BY paired_at DESC`, tid)
	if err != nil {
		return []store.PairedDeviceData{}
	}

	result := make([]store.PairedDeviceData, len(rows))
	for i, r := range rows {
		result[i] = store.PairedDeviceData{
			SenderID: r.SenderID, Channel: r.Channel, ChatID: r.ChatID,
			PairedBy: r.PairedBy, PairedAt: r.PairedAt.UnixMilli(),
		}
		if r.ExpiresAt != nil {
			ms := r.ExpiresAt.UnixMilli()
			result[i].ExpiresAt = &ms
		}
		if len(r.Metadata) > 0 {
			json.Unmarshal(r.Metadata, &result[i].Metadata)
		}
	}
	return result
}

func (s *PGPairingStore) MigrateGroupChatID(ctx context.Context, channel, oldChatID, newChatID string) error {
	tid := tenantIDForInsert(ctx)

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin migrate tx: %w", err)
	}
	defer tx.Rollback()

	// 1. paired_devices: update sender_id and chat_id
	if _, err := tx.ExecContext(ctx,
		`UPDATE paired_devices
		 SET sender_id = REPLACE(sender_id, $1, $2),
		     chat_id = REPLACE(chat_id, $1, $2)
		 WHERE sender_id LIKE '%' || $1 || '%'
		   AND channel = $3
		   AND tenant_id = $4`,
		oldChatID, newChatID, channel, tid,
	); err != nil {
		return fmt.Errorf("migrate paired_devices: %w", err)
	}

	// 2. sessions: update session_key and user_id
	if _, err := tx.ExecContext(ctx,
		`UPDATE sessions
		 SET session_key = REPLACE(session_key, ':' || $1, ':' || $2),
		     user_id = REPLACE(user_id, ':' || $1, ':' || $2)
		 WHERE session_key LIKE '%:telegram:%:' || $1 || '%'
		   AND tenant_id = $3`,
		oldChatID, newChatID, tid,
	); err != nil {
		return fmt.Errorf("migrate sessions: %w", err)
	}

	// 3. channel_contacts: update sender_id
	if _, err := tx.ExecContext(ctx,
		`UPDATE channel_contacts
		 SET sender_id = REPLACE(sender_id, $1, $2)
		 WHERE sender_id LIKE '%' || $1 || '%'
		   AND channel_type = 'telegram'
		   AND tenant_id = $3`,
		oldChatID, newChatID, tid,
	); err != nil {
		return fmt.Errorf("migrate channel_contacts: %w", err)
	}

	// 4. channel_pending_messages: update history_key
	if _, err := tx.ExecContext(ctx,
		`UPDATE channel_pending_messages
		 SET history_key = REPLACE(history_key, $1, $2)
		 WHERE history_key LIKE '%' || $1 || '%'
		   AND channel_name = $3
		   AND tenant_id = $4`,
		oldChatID, newChatID, channel, tid,
	); err != nil {
		return fmt.Errorf("migrate channel_pending_messages: %w", err)
	}

	return tx.Commit()
}

func generatePairingCode() string {
	b := make([]byte, codeLength)
	rand.Read(b)
	code := make([]byte, codeLength)
	for i := range code {
		code[i] = codeAlphabet[int(b[i])%len(codeAlphabet)]
	}
	return string(code)
}