package im

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
)

const (
	streamEventInsertArgCount  = 14
	streamEventInsertBatchSize = 200
)

func numberedPlaceholders(start int, count int) string {
	placeholders := make([]string, 0, count)
	for idx := 0; idx < count; idx++ {
		placeholders = append(placeholders, fmt.Sprintf("$%d", start+idx))
	}
	return "(" + strings.Join(placeholders, ", ") + ")"
}

// AppendEvents appends events to the member-local stream. Events with EventSeq
// unset are assigned monotonically within (workspace_id, member_id).
func (store *Store) AppendEvents(ctx context.Context, workspaceID string, memberID string, events []StreamEvent) error {
	if len(events) == 0 {
		return nil
	}
	if err := store.Validate(); err != nil {
		return err
	}
	workspaceID, memberID, err := requireWorkspaceMember(workspaceID, memberID)
	if err != nil {
		return err
	}
	return coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		nextSeq, err := nextEventSeq(ctx, tx, workspaceID, memberID)
		if err != nil {
			return err
		}
		for _, event := range events {
			event.WorkspaceID = workspaceID
			event.MemberID = memberID
			if event.EventSeq <= 0 {
				event.EventSeq = nextSeq
				nextSeq++
			} else if event.EventSeq >= nextSeq {
				nextSeq = event.EventSeq + 1
			}
			if _, err := insertStreamEvent(ctx, tx, event); err != nil {
				return err
			}
		}
		return nil
	})
}

func nextEventSeq(ctx context.Context, queryer coredb.Queryer, workspaceID string, memberID string) (int64, error) {
	row := queryer.QueryRow(ctx, `
SELECT event_seq
FROM im_stream_events
WHERE workspace_id = $1
  AND member_id = $2
ORDER BY event_seq DESC
LIMIT 1
FOR UPDATE
`, workspaceID, memberID)
	var latest int64
	if err := row.Scan(&latest); err != nil {
		if coredb.IsNotFound(err) {
			return 1, nil
		}
		return 0, fmt.Errorf("load next im stream event seq: %w", err)
	}
	if latest < 0 {
		return 1, nil
	}
	return latest + 1, nil
}

func nextEventSeqs(ctx context.Context, queryer coredb.Queryer, workspaceID string, memberIDs []string) (map[string]int64, error) {
	nextByMember := make(map[string]int64, len(memberIDs))
	if len(memberIDs) == 0 {
		return nextByMember, nil
	}
	workspaceID = normalizedString(workspaceID)
	if workspaceID == "" {
		return nil, fmt.Errorf("workspace id is required")
	}
	normalizedMemberIDs := make([]string, 0, len(memberIDs))
	for _, memberID := range memberIDs {
		memberID = normalizedString(memberID)
		if memberID == "" {
			return nil, fmt.Errorf("member id is required")
		}
		if _, ok := nextByMember[memberID]; ok {
			continue
		}
		nextByMember[memberID] = 1
		normalizedMemberIDs = append(normalizedMemberIDs, memberID)
	}
	if len(normalizedMemberIDs) == 0 {
		return nextByMember, nil
	}

	args := make([]any, 0, len(normalizedMemberIDs)+1)
	args = append(args, workspaceID)
	placeholders := make([]string, 0, len(normalizedMemberIDs))
	for _, memberID := range normalizedMemberIDs {
		args = append(args, memberID)
		placeholders = append(placeholders, fmt.Sprintf("$%d", len(args)))
	}
	rows, err := queryer.Query(ctx, fmt.Sprintf(`
SELECT stream.member_id, stream.event_seq
FROM im_stream_events stream
JOIN (
  SELECT member_id, MAX(event_seq) AS event_seq
  FROM im_stream_events
  WHERE workspace_id = $1
    AND member_id IN (%s)
  GROUP BY member_id
) latest
  ON latest.member_id = stream.member_id
 AND latest.event_seq = stream.event_seq
WHERE stream.workspace_id = $1
FOR UPDATE
`, strings.Join(placeholders, ", ")), args...)
	if err != nil {
		return nil, fmt.Errorf("load next im stream event seqs: %w", err)
	}
	defer rows.Close()

	for rows.Next() {
		var memberID string
		var latest int64
		if err := rows.Scan(&memberID, &latest); err != nil {
			return nil, fmt.Errorf("scan next im stream event seq: %w", err)
		}
		memberID = normalizedString(memberID)
		if latest < 0 {
			nextByMember[memberID] = 1
			continue
		}
		nextByMember[memberID] = latest + 1
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate next im stream event seqs: %w", err)
	}
	return nextByMember, nil
}

func insertStreamEvent(ctx context.Context, execer coredb.QueryExecer, event StreamEvent) (bool, error) {
	args, err := streamEventInsertArgs(event)
	if err != nil {
		return false, err
	}
	_, err = execer.Exec(ctx, `
INSERT INTO im_stream_events (
  workspace_id, member_id, event_seq, event_id, type, scope_id,
  conversation_id, entity_id, entity_seq, payload_json, patches_json,
  cursor_generation, created_at, expires_at
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
`, args...)
	if err != nil {
		if uniqueViolationMatches(err, "im_stream_events_event_id_key") {
			return false, nil
		}
		return false, fmt.Errorf("insert im stream event: %w", err)
	}
	return true, nil
}

func uniqueViolationMatches(err error, constraints ...string) bool {
	constraint, ok := coredb.UniqueViolationConstraint(err)
	if !ok {
		return false
	}
	for _, expected := range constraints {
		if constraint == expected || strings.HasSuffix(constraint, "."+expected) {
			return true
		}
	}
	return false
}

func insertStreamEvents(ctx context.Context, execer coredb.QueryExecer, events []StreamEvent) ([]StreamEvent, error) {
	inserted := make([]StreamEvent, 0, len(events))
	for start := 0; start < len(events); start += streamEventInsertBatchSize {
		end := start + streamEventInsertBatchSize
		if end > len(events) {
			end = len(events)
		}
		chunkInserted, err := insertStreamEventBatch(ctx, execer, events[start:end])
		if err != nil {
			return nil, err
		}
		inserted = append(inserted, chunkInserted...)
	}
	return inserted, nil
}

func insertStreamEventBatch(ctx context.Context, execer coredb.QueryExecer, events []StreamEvent) ([]StreamEvent, error) {
	if len(events) == 0 {
		return nil, nil
	}
	args := make([]any, 0, len(events)*streamEventInsertArgCount)
	values := make([]string, 0, len(events))
	for _, event := range events {
		eventArgs, err := streamEventInsertArgs(event)
		if err != nil {
			return nil, err
		}
		args = append(args, eventArgs...)
		values = append(values, numberedPlaceholders(len(args)-streamEventInsertArgCount+1, streamEventInsertArgCount))
	}
	_, err := execer.Exec(ctx, fmt.Sprintf(`
INSERT INTO im_stream_events (
  workspace_id, member_id, event_seq, event_id, type, scope_id,
  conversation_id, entity_id, entity_seq, payload_json, patches_json,
  cursor_generation, created_at, expires_at
) VALUES %s
`, strings.Join(values, ", ")), args...)
	if err != nil {
		if uniqueViolationMatches(err, "im_stream_events_event_id_key") {
			return insertStreamEventsIndividually(ctx, execer, events)
		}
		return nil, fmt.Errorf("insert im stream events: %w", err)
	}
	return events, nil
}

func insertStreamEventsIndividually(ctx context.Context, execer coredb.QueryExecer, events []StreamEvent) ([]StreamEvent, error) {
	inserted := make([]StreamEvent, 0, len(events))
	for _, event := range events {
		ok, err := insertStreamEvent(ctx, execer, event)
		if err != nil {
			return nil, err
		}
		if ok {
			inserted = append(inserted, event)
		}
	}
	return inserted, nil
}

func streamEventInsertArgs(event StreamEvent) ([]any, error) {
	eventID, err := ensureEventID(event.EventID)
	if err != nil {
		return nil, err
	}
	eventType := normalizedString(event.Type)
	if eventType == "" {
		return nil, fmt.Errorf("event type is required")
	}
	payload, err := encodeJSONMap(event.Payload, true)
	if err != nil {
		return nil, err
	}
	patches, err := encodeJSONMap(event.Patches, false)
	if err != nil {
		return nil, err
	}
	createdAt := event.CreatedAt.UTC()
	if createdAt.IsZero() {
		createdAt = nowUTC()
	}
	return []any{
		normalizedString(event.WorkspaceID),
		normalizedString(event.MemberID),
		event.EventSeq,
		eventID,
		eventType,
		normalizedString(event.ScopeID),
		normalizedString(event.ConversationID),
		normalizedString(event.EntityID),
		event.EntitySeq,
		payload,
		patches,
		normalizedString(event.CursorGeneration),
		createdAt,
		event.ExpiresAt,
	}, nil
}

// ListEventsAfter returns a bounded page after afterSeq for WebSocket resume.
func (store *Store) ListEventsAfter(ctx context.Context, workspaceID string, memberID string, afterSeq int64, limit int) ([]StreamEvent, error) {
	if err := store.Validate(); err != nil {
		return nil, err
	}
	workspaceID, memberID, err := requireWorkspaceMember(workspaceID, memberID)
	if err != nil {
		return nil, err
	}
	limit = normalizeLimit(limit, defaultStreamEventListLimit, maxStreamEventListLimit)
	rows, err := store.db.Query(ctx, `
SELECT workspace_id, member_id, event_seq, event_id, type, scope_id,
       conversation_id, entity_id, entity_seq, payload_json, patches_json,
       cursor_generation, created_at, expires_at
FROM im_stream_events
WHERE workspace_id = $1
  AND member_id = $2
  AND event_seq > $3
  AND (expires_at IS NULL OR expires_at > CURRENT_TIMESTAMP)
ORDER BY event_seq ASC
LIMIT $4
`, workspaceID, memberID, afterSeq, limit)
	if err != nil {
		return nil, fmt.Errorf("list im stream events: %w", err)
	}
	defer rows.Close()

	events := make([]StreamEvent, 0)
	for rows.Next() {
		event, err := scanStreamEvent(rows)
		if err != nil {
			return nil, err
		}
		events = append(events, event)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate im stream events: %w", err)
	}
	return events, nil
}

func scanStreamEvent(row interface{ Scan(...any) error }) (StreamEvent, error) {
	var event StreamEvent
	var payloadRaw []byte
	var patchesRaw []byte
	if err := row.Scan(
		&event.WorkspaceID,
		&event.MemberID,
		&event.EventSeq,
		&event.EventID,
		&event.Type,
		&event.ScopeID,
		&event.ConversationID,
		&event.EntityID,
		&event.EntitySeq,
		&payloadRaw,
		&patchesRaw,
		&event.CursorGeneration,
		&event.CreatedAt,
		&event.ExpiresAt,
	); err != nil {
		return StreamEvent{}, fmt.Errorf("scan im stream event: %w", err)
	}
	payload, err := decodeJSONMap(payloadRaw)
	if err != nil {
		return StreamEvent{}, err
	}
	patches, err := decodeJSONMap(patchesRaw)
	if err != nil {
		return StreamEvent{}, err
	}
	event.Payload = payload
	event.Patches = patches
	return event, nil
}

// GetLatestSeq returns the current high-water event seq for a member stream.
func (store *Store) GetLatestSeq(ctx context.Context, workspaceID string, memberID string) (int64, error) {
	if err := store.Validate(); err != nil {
		return 0, err
	}
	workspaceID, memberID, err := requireWorkspaceMember(workspaceID, memberID)
	if err != nil {
		return 0, err
	}
	row := store.db.QueryRow(ctx, `
SELECT COALESCE(MAX(event_seq), 0)
FROM im_stream_events
WHERE workspace_id = $1
  AND member_id = $2
  AND (expires_at IS NULL OR expires_at > CURRENT_TIMESTAMP)
`, workspaceID, memberID)
	var latest int64
	if err := row.Scan(&latest); err != nil {
		return 0, fmt.Errorf("load latest im stream seq: %w", err)
	}
	return latest, nil
}

// DeleteExpiredEvents removes events older than their TTL. It is provided as a
// bounded primitive for future retention workers; first-version API handlers do
// not run background GC.
func (store *Store) DeleteExpiredEvents(ctx context.Context, now time.Time, limit int) (int64, error) {
	if err := store.Validate(); err != nil {
		return 0, err
	}
	limit = normalizeLimit(limit, defaultStreamEventListLimit, maxStreamEventListLimit)
	result, err := store.db.Exec(ctx, `
DELETE FROM im_stream_events
WHERE expires_at IS NOT NULL
  AND expires_at <= $1
LIMIT $2
`, now.UTC(), limit)
	if err != nil {
		return 0, fmt.Errorf("delete expired im stream events: %w", err)
	}
	return result.RowsAffected(), nil
}