go-sip/internal/store/query.go

package store

import (
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"git.ipao.vip/rogee/go-sip/internal/contract"
	"git.ipao.vip/rogee/go-sip/internal/mq"
	"git.ipao.vip/rogee/go-sip/internal/tenant"
	"github.com/google/uuid"
)

var ErrIdempotencyConflict = errors.New("message identity reused with different content")
var ErrMessageScope = errors.New("message dispatcher or tenant binding mismatch")

// HandleQuery saves the query identity and its immutable response in one
// transaction. Returning successfully is sufficient for an input ACK, not for
// claiming that the response has been delivered or consumed.
func (s *Store) HandleQuery(raw []byte, routingKey string) (string, bool, error) {
	request, err := contract.DecodeService(raw)
	if err != nil {
		return "", false, err
	}
	if request.MessageType != "command.query" && request.MessageType != "call.query" {
		return "", false, errors.New("expected command.query or call.query")
	}
	route, err := tenant.NewDispatcherRoute(request.DispatcherID, request.TenantKey)
	if err != nil {
		return "", false, err
	}
	if route.InboundKey != routingKey {
		return "", false, fmt.Errorf("%w: routing mismatch", ErrMessageScope)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	tx, err := s.db.Begin()
	if err != nil {
		return "", false, err
	}
	defer tx.Rollback()
	var owner string
	if err := tx.QueryRow(`SELECT dispatcher_id FROM dispatcher_identity WHERE singleton=1`).Scan(&owner); err != nil {
		return "", false, fmt.Errorf("read query owner: %w", err)
	}
	if owner != request.DispatcherID {
		return "", false, ErrMessageScope
	}
	var conflicting int
	if err := tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM inbox WHERE (tenant_id=? AND tenant_key<>?) OR (tenant_key=? AND tenant_id<>?))`, request.TenantID, request.TenantKey, request.TenantKey, request.TenantID).Scan(&conflicting); err != nil {
		return "", false, err
	}
	if conflicting != 0 {
		return "", false, ErrMessageScope
	}
	if _, err := tx.Exec(`INSERT INTO tenant_bindings(tenant_id,tenant_key) VALUES(?,?) ON CONFLICT DO NOTHING`, request.TenantID, request.TenantKey); err != nil {
		return "", false, err
	}
	var boundKey string
	if err := tx.QueryRow(`SELECT tenant_key FROM tenant_bindings WHERE tenant_id=?`, request.TenantID).Scan(&boundKey); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return "", false, ErrMessageScope
		}
		return "", false, err
	}
	if boundKey != request.TenantKey {
		return "", false, ErrMessageScope
	}
	hash := sha256.Sum256(raw)
	digest := hex.EncodeToString(hash[:])
	var priorHash, responseID string
	err = tx.QueryRow(`SELECT body_hash,response_id FROM mq_query_inbox WHERE tenant_id=? AND message_id=?`, request.TenantID, request.MessageID).Scan(&priorHash, &responseID)
	if err == nil {
		if priorHash != digest {
			return "", false, ErrIdempotencyConflict
		}
		// Re-deliver the persisted reply under its original identity. Never
		// recompute its snapshot or reset an in-flight publication.
		if _, err := tx.Exec(`UPDATE outbox SET status='pending', published_at=NULL WHERE event_id=? AND status='published'`, responseID); err != nil {
			return "", false, err
		}
		if err := tx.Commit(); err != nil {
			return "", false, err
		}
		return responseID, true, nil
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return "", false, err
	}
	var target struct {
		CommandID string `json:"command_id"`
		CallID    string `json:"call_id"`
	}
	if err := json.Unmarshal(request.Payload, &target); err != nil {
		return "", false, err
	}
	now := s.now().UTC()
	deadline, err := time.Parse(time.RFC3339, request.NotAfter)
	if err != nil {
		return "", false, err
	}
	response := contract.ServiceMessage{SchemaVersion: "2.0", MessageType: request.MessageType + ".result", MessageID: uuid.NewString(), DispatcherID: owner, TenantID: request.TenantID, TenantKey: request.TenantKey, TraceID: request.TraceID, IssuedAt: now.Format(time.RFC3339Nano), CorrelationID: request.MessageID, Status: "ok", ReasonCode: "ok"}
	var payload any
	if !now.Before(deadline) {
		response.Status, response.ReasonCode = "rejected", "expired"
		payload = map[string]any{"detail": "query deadline expired", "retryable": false}
	} else {
		if request.MessageType == "command.query" {
			payload, err = commandQuerySnapshot(tx, request.TenantID, request.TenantKey, target.CommandID)
		} else {
			payload, err = callQuerySnapshot(tx, request, target.CallID, now)
		}
		if errors.Is(err, sql.ErrNoRows) {
			response.Status, response.ReasonCode = "rejected", "not_found"
			payload = map[string]any{"detail": "query target not found", "retryable": false}
		} else if errors.Is(err, ErrQuerySnapshotTooLarge) {
			response.Status, response.ReasonCode = "rejected", "unavailable"
			payload = map[string]any{"detail": "query snapshot exceeds MQ message budget; partial results are not returned", "retryable": false}
		} else if err != nil {
			return "", false, err
		}
	}
	response.Payload, err = json.Marshal(payload)
	if err != nil {
		return "", false, err
	}
	body, err := json.Marshal(response)
	if err != nil {
		return "", false, err
	}
	if len(body) > mq.MaxMessageBytes {
		response.Status, response.ReasonCode = "rejected", "unavailable"
		response.Payload = json.RawMessage(`{"detail":"query snapshot exceeds MQ message budget; partial results are not returned","retryable":false}`)
		body, err = json.Marshal(response)
		if err != nil {
			return "", false, err
		}
	}
	if _, err := contract.DecodeService(body); err != nil {
		return "", false, fmt.Errorf("validate query response: %w", err)
	}
	if _, err := tx.Exec(`INSERT INTO outbox(event_id,tenant_key,exchange,routing_key,body,status,created_at) VALUES(?,?,?,?,?,'pending',?)`, response.MessageID, request.TenantKey, mq.EventExchange, route.OutboundKey, body, now.Format(time.RFC3339Nano)); err != nil {
		return "", false, err
	}
	if _, err := tx.Exec(`INSERT INTO mq_query_inbox(tenant_id,message_id,tenant_key,dispatcher_id,body_hash,response_id,received_at) VALUES(?,?,?,?,?,?,?)`, request.TenantID, request.MessageID, request.TenantKey, owner, digest, response.MessageID, now.Format(time.RFC3339Nano)); err != nil {
		return "", false, err
	}
	if err := tx.Commit(); err != nil {
		return "", false, err
	}
	return response.MessageID, false, nil
}

func commandQuerySnapshot(tx *sql.Tx, tenantID, tenantKey, commandID string) (map[string]any, error) {
	var kind, status, received string
	var persisted sql.NullString
	if err := tx.QueryRow(`SELECT command_type,status,received_at,persisted_at FROM inbox WHERE tenant_id=? AND tenant_key=? AND command_id=?`, tenantID, tenantKey, commandID).Scan(&kind, &status, &received, &persisted); err != nil {
		return nil, err
	}
	var version sql.NullInt64
	var reason sql.NullString
	if err := tx.QueryRow(`SELECT json_extract(body,'$.aggregate_version'), json_extract(body,'$.payload.status'),
		json_extract(body,'$.occurred_at'), json_extract(body,'$.payload.reason_code')
		FROM outbox WHERE tenant_key=? AND json_extract(body,'$.tenant_id')=?
		AND json_extract(body,'$.event_type')='command.result'
		AND json_extract(body,'$.aggregate_type')='command' AND json_extract(body,'$.aggregate_id')=?
		ORDER BY json_extract(body,'$.aggregate_version') DESC LIMIT 1`, tenantKey, tenantID, commandID).Scan(&version, &status, &received, &reason); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, errors.New("persisted command has no aggregate-state evidence")
		}
		return nil, err
	}
	if !version.Valid || version.Int64 < 1 || status == "" {
		return nil, errors.New("persisted command has no aggregate-state evidence")
	}
	result := map[string]any{"command_id": commandID, "command_type": kind, "tenant_id": tenantID, "tenant_key": tenantKey, "status": status, "aggregate_version": version.Int64, "updated_at": received}
	if reason.Valid {
		result["reason_code"] = reason.String
	}
	if persisted.Valid {
		result["accepted_at"] = persisted.String
	}
	return result, nil
}