package im

import (
	"context"
	"fmt"
	"strings"

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

// UpdateCursor monotonically advances a client checkpoint using only the
// WebSocket ack path. It refuses to move the checkpoint backwards.
func (store *Store) UpdateCursor(ctx context.Context, workspaceID string, memberID string, clientID string, ackedSeq int64) error {
	if normalizedString(clientID) == "" {
		return fmt.Errorf("client id is required")
	}
	return store.UpdateCursorWithEvent(ctx, workspaceID, memberID, clientID, ackedSeq, "", "")
}

// UpdateCursorWithEvent advances a client checkpoint and stores optional event
// metadata supplied by the ack frame.
func (store *Store) UpdateCursorWithEvent(ctx context.Context, workspaceID string, memberID string, clientID string, ackedSeq int64, ackedEventID string, cursorGeneration string) error {
	if err := store.Validate(); err != nil {
		return err
	}
	workspaceID, memberID, err := requireWorkspaceMember(workspaceID, memberID)
	if err != nil {
		return err
	}
	clientID = normalizedString(clientID)
	if clientID == "" {
		return fmt.Errorf("client id is required")
	}
	if ackedSeq < 0 {
		return fmt.Errorf("acked event seq must not be negative")
	}
	return coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		cursor, err := loadStreamCursor(ctx, tx, workspaceID, memberID, clientID)
		if err != nil {
			return err
		}
		if cursor.ClientID != "" && ackedSeq < cursor.AckedEventSeq {
			return nil
		}
		if _, err := tx.Exec(ctx, `
INSERT INTO im_stream_cursors (
  workspace_id, member_id, client_id, acked_event_seq, acked_event_id,
  cursor_generation, created_at, updated_at
) VALUES ($1, $2, $3, $4, $5, $6, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
ON DUPLICATE KEY UPDATE
  acked_event_seq = GREATEST(acked_event_seq, VALUES(acked_event_seq)),
  acked_event_id = CASE
    WHEN VALUES(acked_event_seq) >= acked_event_seq THEN VALUES(acked_event_id)
    ELSE acked_event_id
  END,
  cursor_generation = CASE
    WHEN VALUES(cursor_generation) <> '' THEN VALUES(cursor_generation)
    ELSE cursor_generation
  END,
  updated_at = CURRENT_TIMESTAMP
`, workspaceID, memberID, clientID, ackedSeq, normalizedString(ackedEventID), normalizedString(cursorGeneration)); err != nil {
			return fmt.Errorf("upsert im stream cursor: %w", err)
		}
		return nil
	})
}

// GetCursor returns a client checkpoint. Missing cursors return a zero-seq
// cursor so callers can bootstrap from snapshot without an extra existence path.
func (store *Store) GetCursor(ctx context.Context, workspaceID string, memberID string, clientID string) (StreamCursor, error) {
	if err := store.Validate(); err != nil {
		return StreamCursor{}, err
	}
	workspaceID, memberID, err := requireWorkspaceMember(workspaceID, memberID)
	if err != nil {
		return StreamCursor{}, err
	}
	clientID = normalizedString(clientID)
	if clientID == "" {
		return StreamCursor{WorkspaceID: workspaceID, MemberID: memberID}, nil
	}
	return loadStreamCursor(ctx, store.db, workspaceID, memberID, clientID)
}

func loadStreamCursor(ctx context.Context, queryer coredb.Queryer, workspaceID string, memberID string, clientID string) (StreamCursor, error) {
	row := queryer.QueryRow(ctx, `
SELECT workspace_id, member_id, client_id, acked_event_seq, acked_event_id,
       cursor_generation, created_at, updated_at
FROM im_stream_cursors
WHERE workspace_id = $1
  AND member_id = $2
  AND client_id = $3
`, workspaceID, memberID, clientID)
	var cursor StreamCursor
	if err := row.Scan(
		&cursor.WorkspaceID,
		&cursor.MemberID,
		&cursor.ClientID,
		&cursor.AckedEventSeq,
		&cursor.AckedEventID,
		&cursor.CursorGeneration,
		&cursor.CreatedAt,
		&cursor.UpdatedAt,
	); err != nil {
		if coredb.IsNotFound(err) {
			return StreamCursor{WorkspaceID: workspaceID, MemberID: memberID, ClientID: clientID}, nil
		}
		return StreamCursor{}, fmt.Errorf("load im stream cursor: %w", err)
	}
	return cursor, nil
}

// ListCursors returns active client checkpoints for diagnostics and future
// retention workers.
func (store *Store) ListCursors(ctx context.Context, workspaceID string, memberID string, limit int) ([]StreamCursor, 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, defaultConversationIndexListSize, maxConversationIndexListSize)
	rows, err := store.db.Query(ctx, `
SELECT workspace_id, member_id, client_id, acked_event_seq, acked_event_id,
       cursor_generation, created_at, updated_at
FROM im_stream_cursors
WHERE workspace_id = $1
  AND member_id = $2
ORDER BY updated_at DESC, client_id ASC
LIMIT $3
`, workspaceID, memberID, limit)
	if err != nil {
		return nil, fmt.Errorf("list im stream cursors: %w", err)
	}
	defer rows.Close()

	items := make([]StreamCursor, 0)
	for rows.Next() {
		var cursor StreamCursor
		if err := rows.Scan(
			&cursor.WorkspaceID,
			&cursor.MemberID,
			&cursor.ClientID,
			&cursor.AckedEventSeq,
			&cursor.AckedEventID,
			&cursor.CursorGeneration,
			&cursor.CreatedAt,
			&cursor.UpdatedAt,
		); err != nil {
			return nil, fmt.Errorf("scan im stream cursor: %w", err)
		}
		cursor.ClientID = strings.TrimSpace(cursor.ClientID)
		items = append(items, cursor)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate im stream cursors: %w", err)
	}
	return items, nil
}