package worker

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"os"
	"time"

	agentdom "github.com/Paca-AI/api/internal/domain/agent"
	notificationdom "github.com/Paca-AI/api/internal/domain/notification"
	projectdom "github.com/Paca-AI/api/internal/domain/project"
	taskdom "github.com/Paca-AI/api/internal/domain/task"
	"github.com/Paca-AI/api/internal/events"
	"github.com/google/uuid"
	"github.com/redis/go-redis/v9"
)

const (
	notificationConsumerGroup = "api.notification_writer"
	notificationReadBlock     = 5 * time.Second
	notificationReadCount     = 50
)

// memberReader resolves project_members rows.
type memberReader interface {
	FindMemberByID(ctx context.Context, memberID uuid.UUID) (*projectdom.ProjectMember, error)
	FindMemberByUserProject(ctx context.Context, userID, projectID uuid.UUID) (*projectdom.ProjectMember, error)
}

// agentTaskTrigger creates an agent conversation when a task is assigned to an agent member.
type agentTaskTrigger interface {
	TriggerTaskAssigned(ctx context.Context, projectID, agentID, taskID, triggeredByMemberID uuid.UUID) (*agentdom.AgentConversation, error)
}

// agentActivityRecorder posts system-generated task activities.
type agentActivityRecorder interface {
	RecordActivity(ctx context.Context, in taskdom.RecordActivityInput) error
}

// NotificationConsumer reads task-assignment events from StreamTaskAssignments
// and creates in-app notifications via the notification service.
//
// The notification service internally resolves member IDs, prevents
// self-notifications, and publishes real-time events to the Valkey Pub/Sub
// channel so the realtime service can push them to connected clients.
type NotificationConsumer struct {
	client          *redis.Client
	notificationSvc notificationdom.Service
	memberRepo      memberReader
	agentSvc        agentTaskTrigger
	activityRec     agentActivityRecorder
	log             *slog.Logger
	consumerName    string
	stopCh          chan struct{}
	doneCh          chan struct{}
}

// NewNotificationConsumer creates a consumer that is ready to be started.
// memberRepo and agentSvc may be nil; agent triggering is then skipped.
func NewNotificationConsumer(client *redis.Client, notificationSvc notificationdom.Service, log *slog.Logger, memberRepo memberReader, agentSvc agentTaskTrigger) *NotificationConsumer {
	hostname, err := os.Hostname()
	if err != nil || hostname == "" {
		hostname = uuid.New().String()
	}
	return &NotificationConsumer{
		client:          client,
		notificationSvc: notificationSvc,
		memberRepo:      memberRepo,
		agentSvc:        agentSvc,
		log:             log,
		consumerName:    fmt.Sprintf("%s.%s", notificationConsumerGroup, hostname),
		stopCh:          make(chan struct{}),
		doneCh:          make(chan struct{}),
	}
}

// WithActivityRecorder attaches an activity recorder so that an
// "agent.session.started" activity is appended to the task's history
// each time an agent conversation is triggered.
func (c *NotificationConsumer) WithActivityRecorder(r agentActivityRecorder) *NotificationConsumer {
	c.activityRec = r
	return c
}

// Start creates the consumer group if needed and begins processing in a
// background goroutine. Call Stop to drain and exit cleanly.
func (c *NotificationConsumer) Start(ctx context.Context) {
	err := c.client.XGroupCreateMkStream(ctx, events.StreamTaskAssignments, notificationConsumerGroup, "0").Err()
	if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
		c.log.Warn("notification consumer: could not create consumer group", "err", err)
	}

	go c.run()
}

// Stop signals the consumer to stop and waits for the goroutine to exit.
func (c *NotificationConsumer) Stop() {
	close(c.stopCh)
	<-c.doneCh
}

func (c *NotificationConsumer) run() {
	defer close(c.doneCh)
	c.log.Info("notification consumer: started", "stream", events.StreamTaskAssignments)

	c.processPending(context.Background())

	for {
		select {
		case <-c.stopCh:
			c.log.Info("notification consumer: stopping")
			return
		default:
		}

		ctx, cancel := context.WithTimeout(context.Background(), notificationReadBlock+time.Second)
		msgs, err := c.client.XReadGroup(ctx, &redis.XReadGroupArgs{
			Group:    notificationConsumerGroup,
			Consumer: c.consumerName,
			Streams:  []string{events.StreamTaskAssignments, ">"},
			Count:    notificationReadCount,
			Block:    notificationReadBlock,
		}).Result()
		cancel()

		if err != nil {
			if err == redis.Nil {
				continue
			}
			c.log.Error("notification consumer: xreadgroup error", "err", err)
			time.Sleep(2 * time.Second)
			continue
		}

		for _, stream := range msgs {
			for _, msg := range stream.Messages {
				c.handle(msg)
			}
		}
	}
}

func (c *NotificationConsumer) processPending(ctx context.Context) {
	msgs, err := c.client.XReadGroup(ctx, &redis.XReadGroupArgs{
		Group:    notificationConsumerGroup,
		Consumer: c.consumerName,
		Streams:  []string{events.StreamTaskAssignments, "0"},
		Count:    notificationReadCount,
	}).Result()
	if err != nil && err != redis.Nil {
		c.log.Warn("notification consumer: could not read pending messages", "err", err)
		return
	}
	for _, stream := range msgs {
		for _, msg := range stream.Messages {
			c.handle(msg)
		}
	}
}

// assignmentStreamPayload mirrors the JSON shape produced by the task handler
// when it appends to StreamTaskAssignments.
type assignmentStreamPayload struct {
	TaskID              string `json:"task_id"`
	ProjectID           string `json:"project_id"`
	NewAssigneeMemberID string `json:"new_assignee_member_id"`
	OldAssigneeMemberID string `json:"old_assignee_member_id,omitempty"`
	ActorUserID         string `json:"actor_user_id"`
}

func (c *NotificationConsumer) handle(msg redis.XMessage) {
	ctx := context.Background()

	raw, ok := msg.Values["payload"].(string)
	if !ok {
		c.log.Warn("notification consumer: message has no payload field", "id", msg.ID)
		c.ack(ctx, msg.ID)
		return
	}

	var p assignmentStreamPayload
	if err := json.Unmarshal([]byte(raw), &p); err != nil {
		c.log.Warn("notification consumer: failed to decode payload", "id", msg.ID, "err", err)
		c.ack(ctx, msg.ID)
		return
	}

	taskID, err := uuid.Parse(p.TaskID)
	if err != nil {
		c.log.Warn("notification consumer: invalid task_id", "id", msg.ID, "task_id", p.TaskID)
		c.ack(ctx, msg.ID)
		return
	}
	projectID, err := uuid.Parse(p.ProjectID)
	if err != nil {
		c.log.Warn("notification consumer: invalid project_id", "id", msg.ID)
		c.ack(ctx, msg.ID)
		return
	}
	newAssigneeMemberID, err := uuid.Parse(p.NewAssigneeMemberID)
	if err != nil {
		c.log.Warn("notification consumer: invalid new_assignee_member_id", "id", msg.ID)
		c.ack(ctx, msg.ID)
		return
	}
	actorUserID, err := uuid.Parse(p.ActorUserID)
	if err != nil {
		c.log.Warn("notification consumer: invalid actor_user_id", "id", msg.ID)
		c.ack(ctx, msg.ID)
		return
	}

	// Determine whether the assignee is an agent member.
	// Agent members have no user_id, so sending them a notification would
	// violate the FK constraint on notifications.recipient_user_id. Instead,
	// we publish a trigger to the AI agent and skip the human notification.
	if c.memberRepo != nil && c.agentSvc != nil {
		member, err := c.memberRepo.FindMemberByID(ctx, newAssigneeMemberID)
		if err == nil && member.IsAgent() && member.AgentID != nil {
			var actorMemberID uuid.UUID
			if actorMember, err := c.memberRepo.FindMemberByUserProject(ctx, actorUserID, projectID); err == nil {
				actorMemberID = actorMember.ID
			}
			conv, triggerErr := c.agentSvc.TriggerTaskAssigned(ctx, projectID, *member.AgentID, taskID, actorMemberID)
			if triggerErr != nil {
				c.log.Error("notification consumer: TriggerTaskAssigned failed", "id", msg.ID, "err", triggerErr)
			} else if conv != nil && c.activityRec != nil {
				content, _ := json.Marshal(map[string]any{
					"conversation_id": conv.ID.String(),
					"agent_id":        member.AgentID.String(),
				})
				agentID := *member.AgentID
				if recErr := c.activityRec.RecordActivity(ctx, taskdom.RecordActivityInput{
					TaskID:       taskID,
					ProjectID:    projectID,
					ActorAgentID: &agentID,
					ActivityType: taskdom.ActivityTypeAgentSessionStarted,
					Content:      content,
				}); recErr != nil {
					c.log.Warn("notification consumer: could not record agent session activity", "err", recErr)
				}
			}
			c.ack(ctx, msg.ID)
			return
		}
	}

	if err := c.notificationSvc.NotifyAssigned(ctx, notificationdom.NotifyAssignedInput{
		TaskID:              taskID,
		ProjectID:           projectID,
		NewAssigneeMemberID: newAssigneeMemberID,
		ActorUserID:         actorUserID,
	}); err != nil {
		c.log.Error("notification consumer: NotifyAssigned failed", "id", msg.ID, "err", err)
		// Do not ack — will be retried via processPending on next restart.
		return
	}

	c.ack(ctx, msg.ID)
}

func (c *NotificationConsumer) ack(ctx context.Context, id string) {
	if err := c.client.XAck(ctx, events.StreamTaskAssignments, notificationConsumerGroup, id).Err(); err != nil {
		c.log.Warn("notification consumer: xack failed", "id", id, "err", err)
	}
}
