package notification

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgxpool"
)

var ErrNotAvailable = errors.New("notification: service not available")

// Service Outbox 与站内通知。
type Service struct {
	pool *pgxpool.Pool
}

func NewService(pool *pgxpool.Pool) *Service {
	return &Service{pool: pool}
}

func (s *Service) Available() bool {
	if s == nil || s.pool == nil {
		return false
	}
	var ok bool
	_ = s.pool.QueryRow(context.Background(), `
		SELECT EXISTS (
			SELECT 1 FROM information_schema.tables
			WHERE table_schema = 'public' AND table_name = 'outbox_events'
		)
	`).Scan(&ok)
	return ok
}

// EmitStageReadyToAdvance 写入 Outbox 与创建者通知；去重键 (stage_instance_id, readiness_revision)。
func (s *Service) EmitStageReadyToAdvance(ctx context.Context, creatorUserID int64, payload StageReadyPayload) (bool, error) {
	if !s.Available() {
		return false, ErrNotAvailable
	}
	if creatorUserID <= 0 || payload.StageInstanceID <= 0 {
		return false, nil
	}

	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return false, err
	}
	defer tx.Rollback(ctx)

	var advanceID int64
	err = tx.QueryRow(ctx, `
		INSERT INTO stage_advance_requests (
			stage_instance_id, workflow_instance_id, stage_key, readiness_revision, status
		) VALUES ($1, $2, $3, $4, 'pending')
		ON CONFLICT (stage_instance_id, readiness_revision) DO NOTHING
		RETURNING id
	`, payload.StageInstanceID, payload.WorkflowInstanceID, payload.StageKey, payload.ConfigRevision).Scan(&advanceID)
	if errors.Is(err, pgx.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}

	raw, err := json.Marshal(payload)
	if err != nil {
		return false, err
	}

	var eventID int64
	aggregateID := fmt.Sprintf("stage_instance:%d", payload.StageInstanceID)
	err = tx.QueryRow(ctx, `
		INSERT INTO outbox_events (aggregate_type, aggregate_id, event_type, payload, project_id, published_at)
		VALUES ('stage_instance', $1, $2, $3::jsonb, $4, now())
		RETURNING id
	`, aggregateID, EventStageReadyToAdvance, string(raw), payload.ProjectID).Scan(&eventID)
	if err != nil {
		return false, err
	}

	stageLabel := payload.StageName
	if stageLabel == "" {
		stageLabel = payload.StageKey
	}
	title := fmt.Sprintf("阶段「%s」可推进", stageLabel)
	body := fmt.Sprintf(
		"%d/%d 个必需文档已批准，没有阻断项。请确认是否推进到下一阶段。",
		payload.ApprovedDocCount, payload.RequiredDocCount,
	)
	dedupeKey := fmt.Sprintf("stage_ready:%d:%d", payload.StageInstanceID, payload.ConfigRevision)

	_, err = tx.Exec(ctx, `
		INSERT INTO notifications (user_id, project_id, event_id, type, title, body, dedupe_key, status)
		VALUES ($1, $2, $3, $4, $5, $6, $7, 'unread')
		ON CONFLICT (user_id, dedupe_key) DO NOTHING
	`, creatorUserID, payload.ProjectID, eventID, TypeStageReadyToAdvance, title, body, dedupeKey)
	if err != nil {
		return false, err
	}

	if err := tx.Commit(ctx); err != nil {
		return false, err
	}
	return true, nil
}

// EmitVisibleStageReady writes the unique stage-ready request on the caller's fenced tx.
// Duplicate (stage, revision) or replayed source_event_id is success.
func (s *Service) EmitVisibleStageReady(ctx context.Context, tx pgx.Tx, sourceEventID string, creatorUserID int64, payload StageReadyPayload) error {
	if tx == nil || sourceEventID == "" || creatorUserID <= 0 || payload.StageInstanceID <= 0 {
		return fmt.Errorf("notification: visible stage ready requires source event, creator, and stage")
	}
	var advanceID int64
	err := tx.QueryRow(ctx, `
		INSERT INTO stage_advance_requests (
			stage_instance_id, workflow_instance_id, stage_key, readiness_revision, status, source_event_id
		) VALUES ($1, $2, $3, $4, 'pending', $5)
		ON CONFLICT (stage_instance_id, readiness_revision) DO NOTHING
		RETURNING id
	`, payload.StageInstanceID, payload.WorkflowInstanceID, payload.StageKey, payload.ConfigRevision, sourceEventID).Scan(&advanceID)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil
	}
	if err != nil {
		return ignoreUnique(err)
	}

	raw, err := json.Marshal(payload)
	if err != nil {
		return err
	}
	var eventID int64
	aggregateID := fmt.Sprintf("stage_instance:%d", payload.StageInstanceID)
	if err := tx.QueryRow(ctx, `
		INSERT INTO outbox_events (aggregate_type, aggregate_id, event_type, payload, project_id, published_at)
		VALUES ('stage_instance', $1, $2, $3::jsonb, $4, now())
		RETURNING id
	`, aggregateID, EventStageReadyToAdvance, string(raw), payload.ProjectID).Scan(&eventID); err != nil {
		return err
	}
	stageLabel := payload.StageName
	if stageLabel == "" {
		stageLabel = payload.StageKey
	}
	title := fmt.Sprintf("阶段「%s」可推进", stageLabel)
	body := fmt.Sprintf(
		"%d/%d 个必需文档已批准，没有阻断项。请确认是否推进到下一阶段。",
		payload.ApprovedDocCount, payload.RequiredDocCount,
	)
	dedupeKey := fmt.Sprintf("stage_ready:%d:%d", payload.StageInstanceID, payload.ConfigRevision)
	_, err = tx.Exec(ctx, `
		INSERT INTO notifications (user_id, project_id, event_id, type, title, body, dedupe_key, status, source_event_id)
		VALUES ($1,$2,$3,$4,$5,$6,$7,'unread',$8)
		ON CONFLICT (user_id, dedupe_key) DO NOTHING
	`, creatorUserID, payload.ProjectID, eventID, TypeStageReadyToAdvance, title, body, dedupeKey, sourceEventID)
	return ignoreUnique(err)
}

func ignoreUnique(err error) error {
	if err == nil {
		return nil
	}
	var pgErr *pgconn.PgError
	if errors.As(err, &pgErr) && pgErr.Code == "23505" {
		return nil
	}
	return err
}

// ListProjectNotifications 列出项目下当前用户的通知（可选 since_id 增量）。
func (s *Service) ListProjectNotifications(ctx context.Context, userID, projectID int64, sinceID int64, limit int) ([]Notification, error) {
	return s.listNotifications(ctx, userID, projectID, sinceID, limit, false)
}

// ListUserNotifications 列出当前用户跨项目通知；action_required 优先，再按 id 降序。
func (s *Service) ListUserNotifications(ctx context.Context, userID, sinceID int64, limit int) ([]Notification, error) {
	return s.listNotifications(ctx, userID, 0, sinceID, limit, true)
}

func (s *Service) listNotifications(ctx context.Context, userID, projectID, sinceID int64, limit int, allProjects bool) ([]Notification, error) {
	if !s.Available() {
		return nil, ErrNotAvailable
	}
	if limit <= 0 || limit > 100 {
		limit = 50
	}

	query := `
		SELECT n.id, n.user_id, n.project_id, COALESCE(n.event_id, 0), n.type, n.title, n.body, n.status,
		       COALESCE(n.severity, 'info'), COALESCE(n.digest_id, 0), n.created_at,
		       COALESCE(e.payload, '{}'::jsonb),
		       COALESCE(d.doc_id, ''),
		       COALESCE((
		         SELECT elem->>'proposal_id'
		         FROM jsonb_array_elements(COALESCE(d.item_summaries, '[]'::jsonb)) WITH ORDINALITY AS t(elem, ord)
		         WHERE COALESCE(elem->>'proposal_id', '') <> ''
		         ORDER BY ord DESC
		         LIMIT 1
		       ), '')
		FROM notifications n
		LEFT JOIN outbox_events e ON e.id = n.event_id
		LEFT JOIN notification_digests d ON d.id = n.digest_id
		WHERE n.user_id = $1
	`
	args := []any{userID}
	if !allProjects {
		query += ` AND n.project_id = $2`
		args = append(args, projectID)
	}
	if sinceID > 0 {
		query += ` AND n.id > $` + fmt.Sprint(len(args)+1)
		args = append(args, sinceID)
	}
	if allProjects {
		query += ` ORDER BY CASE WHEN COALESCE(n.severity, 'info') = 'action_required' THEN 0 ELSE 1 END, n.id DESC`
	} else {
		query += ` ORDER BY n.id ASC`
	}
	query += ` LIMIT $` + fmt.Sprint(len(args)+1)
	args = append(args, limit)

	rows, err := s.pool.Query(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	out := make([]Notification, 0)
	for rows.Next() {
		var n Notification
		var payloadRaw []byte
		var digestDocID, digestProposalID string
		if err := rows.Scan(
			&n.ID, &n.UserID, &n.ProjectID, &n.EventID, &n.Type, &n.Title, &n.Body, &n.Status,
			&n.Severity, &n.DigestID, &n.CreatedAt,
			&payloadRaw, &digestDocID, &digestProposalID,
		); err != nil {
			return nil, err
		}
		if len(payloadRaw) > 0 {
			_ = json.Unmarshal(payloadRaw, &n.Payload)
		}
		if n.Payload.DocID == "" && digestDocID != "" {
			n.Payload.DocID = digestDocID
		}
		if n.Payload.ProposalID == "" && digestProposalID != "" {
			n.Payload.ProposalID = digestProposalID
		}
		out = append(out, n)
	}
	return out, rows.Err()
}