multica/server/internal/integrations/channel/engine/resolvers.go

package engine

import (
	"context"
	"errors"

	"github.com/jackc/pgx/v5/pgtype"

	"github.com/multica-ai/multica/server/internal/integrations/channel"
	"github.com/multica-ai/multica/server/internal/service"
	db "github.com/multica-ai/multica/server/pkg/db/generated"
)

// This file defines the pluggable seams the Router runs the inbound pipeline
// through. Everything platform-specific lives behind these interfaces; a
// platform registers a ResolverSet and the channel-agnostic Router stays
// unchanged. The Feishu implementation is the first ResolverSet.

// Outcome categorizes what the Router decided to do with an inbound message.
// Values match the legacy lark outcomes 1:1 so behavior and dashboards carry
// over unchanged.
type Outcome string

const (
	OutcomeDropped       Outcome = "dropped"
	OutcomeNeedsBinding  Outcome = "needs_binding"
	OutcomeIngested      Outcome = "ingested"
	OutcomeAgentOffline  Outcome = "agent_offline"
	OutcomeAgentArchived Outcome = "agent_archived"
)

// DropReason enumerates the drop-audit categories. Values match the legacy
// lark drop reasons 1:1.
type DropReason string

const (
	DropReasonUnboundUser         DropReason = "unbound_user"
	DropReasonNonWorkspaceMember  DropReason = "non_workspace_member"
	DropReasonNotAddressedInGroup DropReason = "not_addressed_in_group"
	DropReasonDuplicate           DropReason = "duplicate"
	DropReasonRevokedInstallation DropReason = "revoked_installation"
	DropReasonInvalidEvent        DropReason = "invalid_event"
)

// Result is the typed verdict the Router produces for one inbound message,
// consumed by the outbound side (OutboundReplier / typing). It mirrors the
// legacy lark.DispatchResult.
type Result struct {
	Outcome        Outcome
	DropReason     DropReason
	InstallationID pgtype.UUID
	ChatSessionID  pgtype.UUID
	// Sender is the platform-native sender id (e.g. Lark open_id), so the
	// replier can target a binding prompt back to the sender.
	Sender          string
	IssueID         pgtype.UUID
	IssueNumber     int32
	IssueIdentifier string
	IssueTitle      string
}

// ResolvedInstallation is the channel-agnostic installation context the Router
// needs after routing. Platform carries the adapter's own installation value
// opaquely so the set's other ports (binder, replier, typing) reuse it without
// a re-fetch; the Router never reads Platform.
type ResolvedInstallation struct {
	ID              pgtype.UUID
	WorkspaceID     pgtype.UUID
	AgentID         pgtype.UUID
	InstallerUserID pgtype.UUID
	Active          bool
	Platform        any
}

// ResolvedIdentity is the sender mapped to a Multica user.
type ResolvedIdentity struct {
	UserID pgtype.UUID
}

// EnsureSessionParams carries the inputs for SessionBinder.EnsureSession.
// Sender is the resolved session creator (the sole human for p2p, the
// installer for group chats — the Router decides which and passes it here).
type EnsureSessionParams struct {
	Installation ResolvedInstallation
	Sender       pgtype.UUID
	Message      channel.InboundMessage
}

// AppendParams carries the inputs for SessionBinder.AppendMessage. ClaimToken
// is the dedup owner-fence token; the binder runs the dedup Mark INSIDE its
// chat_message+session tx so the durable write and the Mark commit atomically.
type AppendParams struct {
	SessionID      pgtype.UUID
	Sender         pgtype.UUID
	InstallationID pgtype.UUID
	Message        channel.InboundMessage
	ClaimToken     pgtype.UUID
}

// AppendResult reports what AppendMessage decided.
type AppendResult struct {
	// IssueCommand is non-nil when the message was an /issue command.
	IssueCommand *IssueCommand
	// DedupMarked is true when AppendMessage finalized the dedup claim in its
	// own tx; the Router then skips the post-pipeline finalize.
	DedupMarked bool
}

// IssueCommand is the parsed /issue command.
type IssueCommand struct {
	Title       string
	Description string
}

// Sentinel errors the resolvers return so the Router can map them to the right
// product outcome instead of an infrastructure failure.
var (
	// ErrInstallationNotFound: no installation matches the message's routing
	// key → invalid_event drop.
	ErrInstallationNotFound = errors.New("engine: installation not found")
	// ErrSenderUnbound: the sender has no identity binding → needs_binding.
	ErrSenderUnbound = errors.New("engine: sender unbound")
	// ErrSenderNotMember: the sender is bound but not a workspace member →
	// non_workspace_member drop.
	ErrSenderNotMember = errors.New("engine: sender not a workspace member")
	// ErrDuplicate: Claim found the message already processed / in flight →
	// duplicate drop.
	ErrDuplicate = errors.New("engine: duplicate message")
	// ErrClaimLost: a concurrent reclaim rotated the dedup token mid-flight →
	// treated as a duplicate.
	ErrClaimLost = errors.New("engine: dedup claim lost")
)

// InstallationResolver routes an inbound message to its installation. The
// adapter reads whatever platform routing key it needs from the message
// (Source or Raw). Return ErrInstallationNotFound when none matches; return a
// ResolvedInstallation with Active=false when it exists but is revoked.
type InstallationResolver interface {
	ResolveInstallation(ctx context.Context, msg channel.InboundMessage) (ResolvedInstallation, error)
}

// IdentityResolver maps the message sender to a Multica user within the
// installation, re-checking workspace membership. Return ErrSenderUnbound or
// ErrSenderNotMember for the product cases.
type IdentityResolver interface {
	ResolveSender(ctx context.Context, inst ResolvedInstallation, msg channel.InboundMessage) (ResolvedIdentity, error)
}

// Deduper is the two-phase idempotency seam. Claim mints an owner-fence token
// (ErrDuplicate when already processed / in flight); Mark/Release are fenced on
// the token (a no-op on token mismatch is not an error).
type Deduper interface {
	Claim(ctx context.Context, installationID pgtype.UUID, messageID string) (claimToken pgtype.UUID, err error)
	Mark(ctx context.Context, installationID pgtype.UUID, messageID string, claimToken pgtype.UUID) error
	Release(ctx context.Context, installationID pgtype.UUID, messageID string, claimToken pgtype.UUID) error
}

// SessionBinder ensures the chat_session and appends the message (with the
// in-tx dedup Mark). AppendMessage returns ErrClaimLost when the token was
// rotated mid-flight.
type SessionBinder interface {
	EnsureSession(ctx context.Context, p EnsureSessionParams) (pgtype.UUID, error)
	AppendMessage(ctx context.Context, p AppendParams) (AppendResult, error)
}

// Auditor records a dropped inbound event (no message body — drop-audit
// policy). instID may be the zero UUID for installation-less events.
type Auditor interface {
	RecordDrop(ctx context.Context, instID pgtype.UUID, msg channel.InboundMessage, reason DropReason) error
}

// OutboundReplier delivers the verdict-driven reply (binding prompt, offline /
// archived notice, /issue confirmation). Optional; nil disables outbound
// replies. Driven off the ACK critical path by the Router.
type OutboundReplier interface {
	Reply(ctx context.Context, inst ResolvedInstallation, msg channel.InboundMessage, res Result)
}

// TypingNotifier shows a "processing" indicator when a message is ingested and
// clears it once the message reaches a terminal outcome. Optional; nil disables
// it.
type TypingNotifier interface {
	// OnIngested shows the indicator for a successfully ingested message.
	OnIngested(ctx context.Context, inst ResolvedInstallation, msg channel.InboundMessage, sessionID pgtype.UUID)
	// OnSettled clears the indicator for a session whose run trigger produced no
	// task (agent offline / archived, or an enqueue failure). In that case no
	// task lifecycle event is ever published, so the platform's own bus-driven
	// clear (on chat-done / task-failed) would never fire and the indicator would
	// stick. The Router calls this from the debounced flush. Idempotent: a
	// session with no indicator is a no-op.
	OnSettled(ctx context.Context, sessionID pgtype.UUID)
}

// ResolverSet is the per-platform bundle the Router runs the pipeline through.
// Installation/Identity/Dedup/Session/Audit are required; Replier/Typing are
// optional. OriginType is the issue.origin_type label written for /issue
// commands from this channel (Feishu: "lark_chat").
type ResolverSet struct {
	Installation InstallationResolver
	Identity     IdentityResolver
	Dedup        Deduper
	Session      SessionBinder
	Audit        Auditor
	Replier      OutboundReplier
	Typing       TypingNotifier
	OriginType   string
}

// IssueCreator is the narrow subset of service.IssueService the Router needs
// for the /issue command. Shared across platforms.
type IssueCreator interface {
	Create(ctx context.Context, p service.IssueCreateParams, opts service.IssueCreateOpts) (service.IssueCreateResult, error)
}

// TaskEnqueuer is the narrow subset of service.TaskService the Router needs to
// trigger a chat run. Shared across platforms.
type TaskEnqueuer interface {
	EnqueueChatTask(ctx context.Context, session db.ChatSession, initiatorUserID pgtype.UUID, forceFreshSession bool) (db.AgentTaskQueue, error)
}

// SessionReader reads the rows the debounced flush + /issue identifier need.
// Shared across platforms; backed by *db.Queries (the channel-backed store).
type SessionReader interface {
	GetChatSession(ctx context.Context, id pgtype.UUID) (db.ChatSession, error)
	GetWorkspace(ctx context.Context, id pgtype.UUID) (db.Workspace, error)
}