// Vikunja is a to-do list application to facilitate your life.
// Copyright 2018-present Vikunja and contributors. All rights reserved.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.

package models

import (
	"encoding/json"
	"fmt"
	"strconv"
	"time"

	"code.vikunja.io/api/pkg/config"
	"code.vikunja.io/api/pkg/db"
	"code.vikunja.io/api/pkg/events"
	"code.vikunja.io/api/pkg/log"
	"code.vikunja.io/api/pkg/metrics"
	"code.vikunja.io/api/pkg/modules/keyvalue"
	"code.vikunja.io/api/pkg/notifications"
	"code.vikunja.io/api/pkg/user"

	"github.com/ThreeDotsLabs/watermill/message"
	"xorm.io/builder"
	"xorm.io/xorm"
)

// RegisterListeners registers all event listeners
func RegisterListeners() {
	if config.MetricsEnabled.GetBool() {
		events.RegisterListener((&ProjectCreatedEvent{}).Name(), &IncreaseProjectCounter{})
		events.RegisterListener((&ProjectDeletedEvent{}).Name(), &DecreaseProjectCounter{})
		events.RegisterListener((&TaskCreatedEvent{}).Name(), &IncreaseTaskCounter{})
		events.RegisterListener((&TaskDeletedEvent{}).Name(), &DecreaseTaskCounter{})
		events.RegisterListener((&TeamDeletedEvent{}).Name(), &DecreaseTeamCounter{})
		events.RegisterListener((&TeamCreatedEvent{}).Name(), &IncreaseTeamCounter{})
		events.RegisterListener((&TaskAttachmentCreatedEvent{}).Name(), &IncreaseAttachmentCounter{})
		events.RegisterListener((&TaskAttachmentDeletedEvent{}).Name(), &DecreaseAttachmentCounter{})
	}
	events.RegisterListener((&TaskCommentCreatedEvent{}).Name(), &SendTaskCommentNotification{})
	events.RegisterListener((&TaskAssigneeCreatedEvent{}).Name(), &SendTaskAssignedNotification{})
	events.RegisterListener((&TaskDeletedEvent{}).Name(), &SendTaskDeletedNotification{})
	events.RegisterListener((&ProjectCreatedEvent{}).Name(), &SendProjectCreatedNotification{})
	events.RegisterListener((&TeamMemberAddedEvent{}).Name(), &SendTeamMemberAddedNotification{})
	events.RegisterListener((&TeamMemberRemovedEvent{}).Name(), &CleanupTaskAssignmentsAfterTeamRemoval{})
	events.RegisterListener((&TaskCommentUpdatedEvent{}).Name(), &HandleTaskCommentEditMentions{})
	events.RegisterListener((&TaskCreatedEvent{}).Name(), &HandleTaskCreateMentions{})
	events.RegisterListener((&TaskUpdatedEvent{}).Name(), &HandleTaskUpdatedMentions{})
	events.RegisterListener((&UserDataExportRequestedEvent{}).Name(), &HandleUserDataExport{})
	events.RegisterListener((&TaskCommentCreatedEvent{}).Name(), &HandleTaskUpdateLastUpdated{})
	events.RegisterListener((&TaskCommentUpdatedEvent{}).Name(), &HandleTaskUpdateLastUpdated{})
	events.RegisterListener((&TaskCommentDeletedEvent{}).Name(), &HandleTaskUpdateLastUpdated{})
	events.RegisterListener((&TaskAssigneeCreatedEvent{}).Name(), &HandleTaskUpdateLastUpdated{})
	events.RegisterListener((&TaskAssigneeDeletedEvent{}).Name(), &HandleTaskUpdateLastUpdated{})
	events.RegisterListener((&TaskAttachmentCreatedEvent{}).Name(), &HandleTaskUpdateLastUpdated{})
	events.RegisterListener((&TaskAttachmentDeletedEvent{}).Name(), &HandleTaskUpdateLastUpdated{})
	events.RegisterListener((&TaskRelationCreatedEvent{}).Name(), &HandleTaskUpdateLastUpdated{})
	events.RegisterListener((&TaskRelationDeletedEvent{}).Name(), &HandleTaskUpdateLastUpdated{})
	events.RegisterListener((&TaskCreatedEvent{}).Name(), &UpdateTaskInSavedFilterViews{})
	events.RegisterListener((&TaskUpdatedEvent{}).Name(), &UpdateTaskInSavedFilterViews{})
	events.RegisterListener((&TaskCommentCreatedEvent{}).Name(), &MarkTaskUnreadOnComment{})
	if config.WebhooksEnabled.GetBool() {
		RegisterEventForWebhook(&TaskCreatedEvent{})
		RegisterEventForWebhook(&TaskUpdatedEvent{})
		RegisterEventForWebhook(&TaskDeletedEvent{})
		RegisterEventForWebhook(&TaskAssigneeCreatedEvent{})
		RegisterEventForWebhook(&TaskAssigneeDeletedEvent{})
		RegisterEventForWebhook(&TaskCommentCreatedEvent{})
		RegisterEventForWebhook(&TaskCommentUpdatedEvent{})
		RegisterEventForWebhook(&TaskCommentDeletedEvent{})
		RegisterEventForWebhook(&TaskAttachmentCreatedEvent{})
		RegisterEventForWebhook(&TaskAttachmentDeletedEvent{})
		RegisterEventForWebhook(&TaskRelationCreatedEvent{})
		RegisterEventForWebhook(&TaskRelationDeletedEvent{})
		RegisterEventForWebhook(&ProjectUpdatedEvent{})
		RegisterEventForWebhook(&ProjectDeletedEvent{})
		RegisterEventForWebhook(&ProjectSharedWithUserEvent{})
		RegisterEventForWebhook(&ProjectSharedWithTeamEvent{})
		RegisterUserDirectedEventForWebhook(&TaskReminderFiredEvent{})
		RegisterUserDirectedEventForWebhook(&TaskOverdueEvent{})
		RegisterUserDirectedEventForWebhook(&TasksOverdueEvent{})

		// Internal delivery listener — one message per webhook with its own retry lifecycle
		events.RegisterListener((&WebhookDeliveryEvent{}).Name(), &WebhookDeliveryListener{})
	}
}

//////
// Task Events

// IncreaseTaskCounter  represents a listener
type IncreaseTaskCounter struct {
}

// Name defines the name for the IncreaseTaskCounter listener
func (s *IncreaseTaskCounter) Name() string {
	return "task.counter.increase"
}

// Handle is executed when the event IncreaseTaskCounter listens on is fired
func (s *IncreaseTaskCounter) Handle(_ *message.Message) (err error) {
	return keyvalue.IncrBy(metrics.TaskCountKey, 1)
}

// DecreaseTaskCounter  represents a listener
type DecreaseTaskCounter struct {
}

// Name defines the name for the DecreaseTaskCounter listener
func (s *DecreaseTaskCounter) Name() string {
	return "task.counter.decrease"
}

// Handle is executed when the event DecreaseTaskCounter listens on is fired
func (s *DecreaseTaskCounter) Handle(_ *message.Message) (err error) {
	return keyvalue.DecrBy(metrics.TaskCountKey, 1)
}

func notifyMentionedUsers(sess *xorm.Session, task *Task, text string, n notifications.NotificationWithSubject) (users map[int64]*user.User, err error) {
	users, err = FindMentionedUsersInText(sess, text)
	if err != nil {
		return
	}

	if len(users) == 0 {
		return
	}

	log.Debugf("Processing %d mentioned users for text %d", len(users), n.SubjectID())

	var notified int
	for _, u := range users {
		can, _, err := task.CanRead(sess, u)
		if err != nil {
			return users, err
		}

		if !can {
			continue
		}

		// Don't notify a user if they were already notified
		dbn, err := notifications.GetNotificationsForNameAndUser(sess, u.ID, n.Name(), n.SubjectID())
		if err != nil {
			return users, err
		}

		if len(dbn) > 0 {
			continue
		}

		err = notifications.Notify(u, n, sess)
		if err != nil {
			return users, err
		}
		notified++
	}

	log.Debugf("Notified %d mentioned users for text %d", notified, n.SubjectID())

	return
}

// SendTaskCommentNotification  represents a listener
type SendTaskCommentNotification struct {
}

// Name defines the name for the SendTaskCommentNotification listener
func (s *SendTaskCommentNotification) Name() string {
	return "task.comment.notification.send"
}

// Handle is executed when the event SendTaskCommentNotification listens on is fired
func (s *SendTaskCommentNotification) Handle(msg *message.Message) (err error) {
	event := &TaskCommentCreatedEvent{}
	err = json.Unmarshal(msg.Payload, event)
	if err != nil {
		return err
	}

	sess := db.NewSession()
	defer sess.Close()

	project, err := GetProjectSimpleByID(sess, event.Task.ProjectID)
	if err != nil {
		return err
	}

	n := &TaskCommentNotification{
		Doer:      event.Doer,
		Task:      event.Task,
		Comment:   event.Comment,
		Mentioned: true,
		Project:   project,
	}
	mentionedUsers, err := notifyMentionedUsers(sess, event.Task, event.Comment.Comment, n)
	if err != nil {
		return err
	}

	subscribers, err := GetSubscriptionsForEntity(sess, SubscriptionEntityTask, event.Task.ID)
	if err != nil {
		return err
	}

	log.Debugf("Sending task comment notifications to %d subscribers for task %d", len(subscribers), event.Task.ID)

	for _, subscriber := range subscribers {
		if subscriber.UserID == event.Doer.ID {
			continue
		}

		if _, has := mentionedUsers[subscriber.UserID]; has {
			continue
		}

		n := &TaskCommentNotification{
			Doer:    event.Doer,
			Task:    event.Task,
			Comment: event.Comment,
			Project: project,
		}
		err = notifications.Notify(subscriber.User, n, sess)
		if err != nil {
			return
		}
	}

	return sess.Commit()
}

// HandleTaskCommentEditMentions  represents a listener
type HandleTaskCommentEditMentions struct {
}

// Name defines the name for the HandleTaskCommentEditMentions listener
func (s *HandleTaskCommentEditMentions) Name() string {
	return "handle.task.comment.edit.mentions"
}

// Handle is executed when the event HandleTaskCommentEditMentions listens on is fired
func (s *HandleTaskCommentEditMentions) Handle(msg *message.Message) (err error) {
	event := &TaskCommentUpdatedEvent{}
	err = json.Unmarshal(msg.Payload, event)
	if err != nil {
		return err
	}

	if event.Task == nil || event.Comment == nil {
		return nil
	}

	sess := db.NewSession()
	defer sess.Close()

	project, err := GetProjectSimpleByID(sess, event.Task.ProjectID)
	if err != nil {
		return err
	}

	n := &TaskCommentNotification{
		Doer:      event.Doer,
		Task:      event.Task,
		Comment:   event.Comment,
		Mentioned: true,
		Project:   project,
	}
	_, err = notifyMentionedUsers(sess, event.Task, event.Comment.Comment, n)
	if err != nil {
		return err
	}
	return sess.Commit()
}

// SendTaskAssignedNotification  represents a listener
type SendTaskAssignedNotification struct {
}

// Name defines the name for the SendTaskAssignedNotification listener
func (s *SendTaskAssignedNotification) Name() string {
	return "task.assigned.notification.send"
}

// Handle is executed when the event SendTaskAssignedNotification listens on is fired
func (s *SendTaskAssignedNotification) Handle(msg *message.Message) (err error) {
	event := &TaskAssigneeCreatedEvent{}
	err = json.Unmarshal(msg.Payload, event)
	if err != nil {
		return err
	}

	sess := db.NewSession()
	defer sess.Close()

	subscribers, err := GetSubscriptionsForEntity(sess, SubscriptionEntityTask, event.Task.ID)
	if err != nil {
		return err
	}

	log.Debugf("Sending task assigned notifications to %d subscribers for task %d", len(subscribers), event.Task.ID)

	task, err := GetTaskByIDSimple(sess, event.Task.ID)
	if err != nil {
		return err
	}

	project, err := GetProjectSimpleByID(sess, task.ProjectID)
	if err != nil {
		return err
	}

	notifiedUsers := make(map[int64]bool)

	for _, subscriber := range subscribers {
		if subscriber.UserID == event.Doer.ID {
			continue
		}

		if notifiedUsers[subscriber.UserID] {
			// Users may be subscribed to the task and the project itself, which leads to double notifications
			continue
		}

		n := &TaskAssignedNotification{
			Doer:     event.Doer,
			Task:     &task,
			Assignee: event.Assignee,
			Target:   subscriber.User,
			Project:  project,
		}
		err = notifications.Notify(subscriber.User, n, sess)
		if err != nil {
			return
		}

		notifiedUsers[subscriber.UserID] = true
	}

	return sess.Commit()
}

// SendTaskDeletedNotification  represents a listener
type SendTaskDeletedNotification struct {
}

// Name defines the name for the SendTaskDeletedNotification listener
func (s *SendTaskDeletedNotification) Name() string {
	return "task.deleted.notification.send"
}

// Handle is executed when the event SendTaskDeletedNotification listens on is fired
func (s *SendTaskDeletedNotification) Handle(msg *message.Message) (err error) {
	event := &TaskDeletedEvent{}
	err = json.Unmarshal(msg.Payload, event)
	if err != nil {
		return err
	}

	sess := db.NewSession()
	defer sess.Close()

	var subscribers []*SubscriptionWithUser
	subscribers, err = GetSubscriptionsForEntity(sess, SubscriptionEntityTask, event.Task.ID)
	// If the task does not exist and no one has explicitly subscribed to it, we won't find any subscriptions for it.
	// Hence, we need to check for subscriptions to the parent project manually.
	if err != nil && (IsErrTaskDoesNotExist(err) || IsErrProjectDoesNotExist(err)) {
		subscribers, err = GetSubscriptionsForEntity(sess, SubscriptionEntityProject, event.Task.ProjectID)
	}
	if err != nil {
		return err
	}

	log.Debugf("Sending task deleted notifications to %d subscribers for task %d", len(subscribers), event.Task.ID)

	for _, subscriber := range subscribers {
		if subscriber.UserID == event.Doer.ID {
			continue
		}

		n := &TaskDeletedNotification{
			Doer: event.Doer,
			Task: event.Task,
		}
		err = notifications.Notify(subscriber.User, n, sess)
		if err != nil {
			return
		}
	}

	return sess.Commit()
}

// HandleTaskCreateMentions  represents a listener
type HandleTaskCreateMentions struct {
}

// Name defines the name for the HandleTaskCreateMentions listener
func (s *HandleTaskCreateMentions) Name() string {
	return "task.created.mentions"
}

// Handle is executed when the event HandleTaskCreateMentions listens on is fired
func (s *HandleTaskCreateMentions) Handle(msg *message.Message) (err error) {
	event := &TaskCreatedEvent{}
	err = json.Unmarshal(msg.Payload, event)
	if err != nil {
		return err
	}

	if event.Task == nil {
		return nil
	}

	sess := db.NewSession()
	defer sess.Close()

	project, err := GetProjectSimpleByID(sess, event.Task.ProjectID)
	if err != nil {
		return err
	}

	n := &UserMentionedInTaskNotification{
		Task:    event.Task,
		Doer:    event.Doer,
		IsNew:   true,
		Project: project,
	}
	_, err = notifyMentionedUsers(sess, event.Task, event.Task.Description, n)
	if err != nil {
		return err
	}
	return sess.Commit()
}

// HandleTaskUpdatedMentions  represents a listener
type HandleTaskUpdatedMentions struct {
}

// Name defines the name for the HandleTaskUpdatedMentions listener
func (s *HandleTaskUpdatedMentions) Name() string {
	return "task.updated.mentions"
}

// Handle is executed when the event HandleTaskUpdatedMentions listens on is fired
func (s *HandleTaskUpdatedMentions) Handle(msg *message.Message) (err error) {
	event := &TaskUpdatedEvent{}
	err = json.Unmarshal(msg.Payload, event)
	if err != nil {
		return err
	}

	if event.Task == nil {
		return nil
	}

	sess := db.NewSession()
	defer sess.Close()

	project, err := GetProjectSimpleByID(sess, event.Task.ProjectID)
	if err != nil {
		return err
	}

	n := &UserMentionedInTaskNotification{
		Task:    event.Task,
		Doer:    event.Doer,
		IsNew:   false,
		Project: project,
	}

	_, err = notifyMentionedUsers(sess, event.Task, event.Task.Description, n)
	if err != nil {
		return err
	}
	return sess.Commit()
}

// HandleTaskUpdateLastUpdated  represents a listener
type HandleTaskUpdateLastUpdated struct {
}

// Name defines the name for the HandleTaskUpdateLastUpdated listener
func (s *HandleTaskUpdateLastUpdated) Name() string {
	return "handle.task.update.last.updated"
}

// Handle is executed when the event HandleTaskUpdateLastUpdated listens on is fired
func (s *HandleTaskUpdateLastUpdated) Handle(msg *message.Message) (err error) {
	// Using a map here allows us to plug this listener to all kinds of task events
	event := map[string]interface{}{}
	err = json.Unmarshal(msg.Payload, &event)
	if err != nil {
		return err
	}

	task, is := event["task"].(map[string]interface{})
	if !is {
		log.Errorf("Event payload does not contain task")
		return
	}

	taskID, is := task["id"]
	if !is {
		log.Errorf("Event payload does not contain a valid task ID")
		return
	}

	var taskIDInt int64
	switch v := taskID.(type) {
	case int64:
		taskIDInt = v
	case int:
		taskIDInt = int64(v)
	case int32:
		taskIDInt = int64(v)
	case float64:
		taskIDInt = int64(v)
	case float32:
		taskIDInt = int64(v)
	default:
		log.Errorf("Event payload does not contain a valid task ID")
		return
	}

	sess := db.NewSession()
	defer sess.Close()

	err = updateTaskLastUpdated(sess, &Task{ID: taskIDInt})
	if err != nil {
		return err
	}

	return sess.Commit()
}

// IncreaseAttachmentCounter  represents a listener
type IncreaseAttachmentCounter struct {
}

// Name defines the name for the IncreaseAttachmentCounter listener
func (s *IncreaseAttachmentCounter) Name() string {
	return "increase.attachment.counter"
}

// Handle is executed when the event IncreaseAttachmentCounter listens on is fired
func (s *IncreaseAttachmentCounter) Handle(_ *message.Message) (err error) {
	return keyvalue.IncrBy(metrics.AttachmentsCountKey, 1)
}

// DecreaseAttachmentCounter  represents a listener
type DecreaseAttachmentCounter struct {
}

// Name defines the name for the DecreaseAttachmentCounter listener
func (s *DecreaseAttachmentCounter) Name() string {
	return "decrease.attachment.counter"
}

// Handle is executed when the event DecreaseAttachmentCounter listens on is fired
func (s *DecreaseAttachmentCounter) Handle(_ *message.Message) (err error) {
	return keyvalue.DecrBy(metrics.AttachmentsCountKey, 1)
}

// UpdateTaskInSavedFilterViews  represents a listener
type UpdateTaskInSavedFilterViews struct {
}

// Name defines the name for the UpdateTaskInSavedFilterViews listener
func (l *UpdateTaskInSavedFilterViews) Name() string {
	return "task.set.saved.filter.views"
}

// Handle is executed when the event UpdateTaskInSavedFilterViews listens on is fired
func (l *UpdateTaskInSavedFilterViews) Handle(msg *message.Message) (err error) {
	event := &TaskCreatedEvent{}
	err = json.Unmarshal(msg.Payload, event)
	if err != nil {
		return err
	}

	if event.Task == nil {
		return nil
	}

	// This operation is potentially very resource-heavy, because we don't know if a task is included
	// in a filter until we evaluate that filter. We need to evaluate each filter individually - since
	// there can be many filters, this can take a while to execute.
	// For this reason, we do this in an asynchronous event listener.

	s := db.NewSession()
	defer s.Close()

	// Get all saved filters with a manual kanban view
	kanbanFilterViews := []*ProjectView{}
	err = s.Where("project_id < 0 and view_kind = ? and bucket_configuration_mode = ?", ProjectViewKindKanban, BucketConfigurationModeManual).
		Find(&kanbanFilterViews)
	if err != nil {
		return err
	}

	filterIDs := []int64{}
	for _, view := range kanbanFilterViews {
		filterIDs = append(filterIDs, GetSavedFilterIDFromProjectID(view.ProjectID))
	}

	filters := map[int64]*SavedFilter{}
	err = s.In("id", filterIDs).Find(&filters)
	if err != nil {
		return err
	}

	var fallbackTimezone string
	if event.Doer != nil {
		u, userErr := user.GetUserByID(s, event.Doer.GetID())
		if userErr == nil {
			fallbackTimezone = u.Timezone
		}
		// When a link share triggered this event, the user id will be 0, and thus this fails.
		// Similarly, when the doer has been deleted, the user will not exist.
		// Only passing the value along when the user was retrieved successfully ensures the whole handler
		// does not fail because of that.
		// When the fallback is empty, it will be handled later anyhow.
	}

	taskBuckets := []*TaskBucket{}
	taskPositions := []*TaskPosition{}

	viewIDToCleanUp := []int64{}

	for _, view := range kanbanFilterViews {
		filter, exists := filters[GetSavedFilterIDFromProjectID(view.ProjectID)]
		if !exists {
			log.Debugf("Did not find filter for view %d", view.ID)
			continue
		}

		taskBucket, taskPosition, err := addTaskToFilter(s, filter, view, fallbackTimezone, event.Task)
		if err != nil {
			if IsErrInvalidFilterExpression(err) ||
				IsErrInvalidTaskFilterValue(err) ||
				IsErrInvalidTaskFilterConcatinator(err) ||
				IsErrInvalidTaskFilterComparator(err) ||
				IsErrInvalidTaskField(err) {
				log.Debugf("Invalid filter expression for view %d, expression: %v", view.ID, view.Filter)
				continue
			}

			return err
		}

		if taskBucket != nil && taskPosition != nil {
			taskBuckets = append(taskBuckets, taskBucket)
			taskPositions = append(taskPositions, taskPosition)
			viewIDToCleanUp = append(viewIDToCleanUp, view.ID)
		}
	}

	if len(taskBuckets) > 0 || len(taskPositions) > 0 {
		_, err = s.And(
			builder.Eq{"task_id": event.Task.ID},
			builder.In("project_view_id", viewIDToCleanUp),
		).
			Delete(&TaskBucket{})
		if err != nil {
			return
		}
		_, err = s.And(
			builder.Eq{"task_id": event.Task.ID},
			builder.In("project_view_id", viewIDToCleanUp),
		).
			Delete(&TaskPosition{})
		if err != nil {
			return
		}
		_, err = s.Insert(taskBuckets)
		if err != nil {
			return
		}
		_, err = s.Insert(taskPositions)
		if err != nil {
			return
		}
	}

	return s.Commit()
}

///////
// Project Event Listeners

type IncreaseProjectCounter struct {
}

func (s *IncreaseProjectCounter) Name() string {
	return "project.counter.increase"
}

func (s *IncreaseProjectCounter) Handle(_ *message.Message) (err error) {
	return keyvalue.IncrBy(metrics.ProjectCountKey, 1)
}

type DecreaseProjectCounter struct {
}

func (s *DecreaseProjectCounter) Name() string {
	return "project.counter.decrease"
}

func (s *DecreaseProjectCounter) Handle(_ *message.Message) (err error) {
	return keyvalue.DecrBy(metrics.ProjectCountKey, 1)
}

// SendProjectCreatedNotification  represents a listener
type SendProjectCreatedNotification struct {
}

// Name defines the name for the SendProjectCreatedNotification listener
func (s *SendProjectCreatedNotification) Name() string {
	return "send.project.created.notification"
}

// Handle is executed when the event SendProjectCreatedNotification listens on is fired
func (s *SendProjectCreatedNotification) Handle(msg *message.Message) (err error) {
	event := &ProjectCreatedEvent{}
	err = json.Unmarshal(msg.Payload, event)
	if err != nil {
		return err
	}

	sess := db.NewSession()
	defer sess.Close()

	subscribers, err := GetSubscriptionsForEntity(sess, SubscriptionEntityProject, event.Project.ID)
	if err != nil {
		return err
	}

	log.Debugf("Sending project created notifications to %d subscribers for project %d", len(subscribers), event.Project.ID)

	for _, subscriber := range subscribers {
		if subscriber.UserID == event.Doer.ID {
			continue
		}

		n := &ProjectCreatedNotification{
			Doer:    event.Doer,
			Project: event.Project,
		}
		err = notifications.Notify(subscriber.User, n, sess)
		if err != nil {
			return
		}
	}

	return sess.Commit()
}

// WebhookListener represents a listener
type WebhookListener struct {
	EventName string
}

// Name defines the name for the WebhookListener listener
func (wl *WebhookListener) Name() string {
	return "webhook.listener"
}

type WebhookPayload struct {
	EventName string      `json:"event_name"`
	Time      time.Time   `json:"time"`
	Data      interface{} `json:"data"`
}

// WebhookDeliveryListener delivers one webhook per message. It is the
// consumer for WebhookDeliveryEvent and owns the retry semantics: any
// error returned from Handle triggers the watermill retry middleware
// independently for this single delivery, with no effect on other
// webhooks on the same parent event.
type WebhookDeliveryListener struct{}

// Name defines the name for the WebhookDeliveryListener listener
func (wdl *WebhookDeliveryListener) Name() string {
	return "webhook.delivery.listener"
}

// Handle is executed when a WebhookDeliveryEvent is fired. It reloads the
// webhook from the database by id (so secrets, target_url, and basic auth
// credentials are always current) and performs the HTTP delivery.
//
// Special cases:
//   - If the webhook row no longer exists (deleted between fan-out and
//     delivery), Handle returns nil so the message is not retried.
//   - A nil payload is treated as data corruption / version skew and
//     returned as an error so the message is retried and eventually
//     parked in the poison queue rather than silently dropped.
//   - Any other error is returned so the watermill retry middleware
//     retries this delivery with exponential backoff, and eventually
//     parks it in the poison queue if all retries fail.
func (wdl *WebhookDeliveryListener) Handle(msg *message.Message) error {
	evt := &WebhookDeliveryEvent{}
	if err := json.Unmarshal(msg.Payload, evt); err != nil {
		return err
	}

	if evt.Payload == nil {
		return fmt.Errorf("webhook delivery event for webhook %d has no payload", evt.WebhookID)
	}

	s := db.NewSession()
	defer s.Close()

	webhook := &Webhook{}
	has, err := s.Where("id = ?", evt.WebhookID).Get(webhook)
	if err != nil {
		return err
	}
	if !has {
		log.Debugf("webhook %d no longer exists, skipping delivery of %s", evt.WebhookID, evt.Payload.EventName)
		return nil
	}

	return webhook.sendWebhookPayload(evt.Payload)
}

func getIDAsInt64(id interface{}) int64 {
	if id == nil {
		return 0
	}

	switch v := id.(type) {
	case int64:
		return v
	case float64:
		return int64(v)
	case float32:
		return int64(v)
	case int:
		return int64(v)
	case int32:
		return int64(v)
	case json.Number:
		if i, err := v.Int64(); err == nil {
			return i
		}
		if f, err := v.Float64(); err == nil {
			return int64(f)
		}
		return 0
	case string:
		if i, err := strconv.ParseInt(v, 10, 64); err == nil {
			return i
		}
		if f, err := strconv.ParseFloat(v, 64); err == nil {
			return int64(f)
		}
		return 0
	default:
		return 0
	}
}

func getProjectIDFromAnyEvent(eventPayload map[string]interface{}) int64 {
	if task, has := eventPayload["task"]; has {
		t := task.(map[string]interface{})
		if projectID, has := t["project_id"]; has {
			return getIDAsInt64(projectID)
		}
	}

	if project, has := eventPayload["project"]; has {
		t := project.(map[string]interface{})
		if projectID, has := t["id"]; has {
			return getIDAsInt64(projectID)
		}
	}

	return 0
}

func getUserIDFromAnyEvent(eventPayload map[string]interface{}) int64 {
	if u, has := eventPayload["user"]; has {
		userMap, ok := u.(map[string]interface{})
		if !ok {
			return 0
		}
		if userID, has := userMap["id"]; has {
			return getIDAsInt64(userID)
		}
	}

	return 0
}

func reloadDoerInEvent(s *xorm.Session, event map[string]interface{}) (doerID int64, err error) {
	doer, has := event["doer"]
	if !has || doer == nil {
		return 0, nil
	}

	// doer can be null in incoming payloads, so guard the type assertion
	d, ok := doer.(map[string]interface{})
	if !ok {
		return 0, nil
	}

	rawDoerID, has := d["id"]
	if !has || rawDoerID == nil {
		return 0, nil
	}

	doerID = getIDAsInt64(rawDoerID)
	if doerID <= 0 {
		return 0, nil
	}

	fullDoer, err := user.GetUserByID(s, doerID)
	if err != nil && !user.IsErrUserDoesNotExist(err) {
		return 0, err
	}
	if err == nil {
		event["doer"] = fullDoer
	}

	return doerID, nil
}

func reloadTaskInEvent(s *xorm.Session, event map[string]interface{}, doerID int64) error {
	task, has := event["task"]
	if !has || task == nil || doerID == 0 {
		return nil
	}

	// guard the type assertion for task as well
	t, ok := task.(map[string]interface{})
	if !ok {
		return nil
	}

	taskID, has := t["id"]
	if !has || taskID == nil {
		return nil
	}

	id := getIDAsInt64(taskID)
	if id <= 0 {
		return nil
	}

	fullTask := Task{
		ID: id,
		Expand: []TaskCollectionExpandable{
			TaskCollectionExpandBuckets,
		},
	}
	err := fullTask.ReadOne(s, &user.User{ID: doerID})
	if err != nil && !IsErrTaskDoesNotExist(err) {
		return err
	}
	if err == nil {
		event["task"] = fullTask
	}

	return nil
}

func reloadProjectInEvent(s *xorm.Session, event map[string]interface{}, projectID, doerID int64) error {
	_, has := event["project"]
	if !has || doerID == 0 {
		return nil
	}

	project, err := GetProjectSimpleByID(s, projectID)
	if err != nil {
		if IsErrProjectDoesNotExist(err) {
			return nil
		}
		return err
	}

	err = project.ReadOne(s, &user.User{ID: doerID})
	if err != nil && !IsErrProjectDoesNotExist(err) {
		return err
	}

	if err == nil {
		event["project"] = project
	}

	return nil
}

func reloadAssigneeInEvent(s *xorm.Session, event map[string]interface{}) error {
	assignee, has := event["assignee"]
	if !has || assignee == nil {
		return nil
	}

	a, ok := assignee.(map[string]interface{})
	if !ok {
		return nil
	}

	assigneeID := getIDAsInt64(a["id"])
	if assigneeID <= 0 {
		return nil
	}

	fullAssignee, err := user.GetUserByID(s, assigneeID)
	if err != nil && !user.IsErrUserDoesNotExist(err) {
		return err
	}
	if err == nil {
		event["assignee"] = fullAssignee
	}

	return nil
}

func reloadUserInEvent(s *xorm.Session, event map[string]interface{}) error {
	u, has := event["user"]
	if !has || u == nil {
		return nil
	}

	userMap, ok := u.(map[string]interface{})
	if !ok {
		return nil
	}

	userID := getIDAsInt64(userMap["id"])
	if userID <= 0 {
		return nil
	}

	fullUser, err := user.GetUserByID(s, userID)
	if err != nil && !user.IsErrUserDoesNotExist(err) {
		return err
	}
	if err == nil {
		event["user"] = fullUser
	}

	return nil
}

func reloadEventData(s *xorm.Session, event map[string]interface{}, projectID int64) (eventWithData map[string]interface{}, doerID int64, err error) {
	// Load event data again so that it is always populated in the webhook payload

	doerID, err = reloadDoerInEvent(s, event)
	if err != nil {
		return nil, 0, err
	}

	err = reloadTaskInEvent(s, event, doerID)
	if err != nil {
		return nil, doerID, err
	}

	err = reloadProjectInEvent(s, event, projectID, doerID)
	if err != nil {
		return nil, doerID, err
	}

	err = reloadAssigneeInEvent(s, event)
	if err != nil {
		return nil, doerID, err
	}

	err = reloadUserInEvent(s, event)
	if err != nil {
		return nil, doerID, err
	}

	return event, doerID, nil
}

// Handle is executed when the event WebhookListener listens on is fired
func (wl *WebhookListener) Handle(msg *message.Message) (err error) {
	var event map[string]interface{}
	err = json.Unmarshal(msg.Payload, &event)
	if err != nil {
		return err
	}

	s := db.NewSession()
	defer s.Close()

	projectID := getProjectIDFromAnyEvent(event)
	isUserDirected := IsUserDirectedEvent(wl.EventName)

	// For non-user-directed events, we need a project ID
	if projectID == 0 && !isUserDirected {
		log.Debugf("event %s does not contain a project id, not handling webhook", wl.EventName)
		return nil
	}

	// Look up project-level webhooks
	matchingWebhooks := []*Webhook{}
	if projectID > 0 {
		parents, err := GetAllParentProjects(s, projectID)
		if err != nil {
			return err
		}

		projectIDs := make([]int64, 0, len(parents)+1)
		projectIDs = append(projectIDs, projectID)
		for _, p := range parents {
			projectIDs = append(projectIDs, p.ID)
		}

		ws := []*Webhook{}
		err = s.In("project_id", projectIDs).
			OrderBy("id ASC").
			Find(&ws)
		if err != nil {
			return err
		}

		for _, w := range ws {
			for _, e := range w.Events {
				if e == wl.EventName {
					matchingWebhooks = append(matchingWebhooks, w)
					break
				}
			}
		}
	}

	// Look up user-level webhooks for user-directed events
	if isUserDirected {
		userID := getUserIDFromAnyEvent(event)
		if userID > 0 {
			userWebhooks := []*Webhook{}
			err = s.Where("user_id = ? AND (project_id IS NULL OR project_id = 0)", userID).
				OrderBy("id ASC").
				Find(&userWebhooks)
			if err != nil {
				return err
			}

			for _, w := range userWebhooks {
				for _, e := range w.Events {
					if e == wl.EventName {
						matchingWebhooks = append(matchingWebhooks, w)
						break
					}
				}
			}
		}
	}

	if len(matchingWebhooks) == 0 {
		log.Debugf("Did not find any webhook for the %s event, not sending", wl.EventName)
		return nil
	}

	var doerID int64
	event, doerID, err = reloadEventData(s, event, projectID)
	if err != nil {
		return err
	}

	now := time.Now()
	for _, webhook := range matchingWebhooks {
		// Clone the event map so each webhook gets its own, isolated payload.
		// Otherwise adding event["project"] for one webhook would leak into
		// payloads dispatched for later webhooks.
		perWebhookEvent := make(map[string]interface{}, len(event)+1)
		for k, v := range event {
			perWebhookEvent[k] = v
		}

		if _, has := perWebhookEvent["project"]; !has && webhook.ProjectID > 0 {
			project, err := GetProjectSimpleByID(s, webhook.ProjectID)
			if err != nil && !IsErrProjectDoesNotExist(err) {
				log.Errorf("Could not load project for webhook %d: %s", webhook.ID, err)
			}
			if project != nil {
				err = project.ReadOne(s, &user.User{ID: doerID})
				if err != nil && !IsErrProjectDoesNotExist(err) {
					log.Errorf("Could not load project for webhook %d: %s", webhook.ID, err)
				}
				if err == nil {
					perWebhookEvent["project"] = project
				}
			}
		}

		dispatchErr := events.Dispatch(&WebhookDeliveryEvent{
			WebhookID: webhook.ID,
			Payload: &WebhookPayload{
				EventName: wl.EventName,
				Time:      now,
				Data:      perWebhookEvent,
			},
		})
		if dispatchErr != nil {
			// A dispatch failure here means the in-process event bus is broken —
			// there is nothing useful to retry per-webhook and we do not want
			// to fail the parent message (which would re-fan-out and duplicate
			// any deliveries that did succeed). Log and move on.
			log.Errorf("Could not dispatch webhook.delivery for webhook %d: %s", webhook.ID, dispatchErr)
		}
	}

	return nil
}

///////
// Team Events

// IncreaseTeamCounter  represents a listener
type IncreaseTeamCounter struct {
}

// Name defines the name for the IncreaseTeamCounter listener
func (s *IncreaseTeamCounter) Name() string {
	return "team.counter.increase"
}

// Handle is executed when the event IncreaseTeamCounter listens on is fired
func (s *IncreaseTeamCounter) Handle(_ *message.Message) (err error) {
	return keyvalue.IncrBy(metrics.TeamCountKey, 1)
}

// DecreaseTeamCounter  represents a listener
type DecreaseTeamCounter struct {
}

// Name defines the name for the DecreaseTeamCounter listener
func (s *DecreaseTeamCounter) Name() string {
	return "team.counter.decrease"
}

// Handle is executed when the event DecreaseTeamCounter listens on is fired
func (s *DecreaseTeamCounter) Handle(_ *message.Message) (err error) {
	return keyvalue.DecrBy(metrics.TeamCountKey, 1)
}

// CleanupTaskAssignmentsAfterTeamRemoval represents a listener
type CleanupTaskAssignmentsAfterTeamRemoval struct{}

// Name defines the name of the listener
func (l *CleanupTaskAssignmentsAfterTeamRemoval) Name() string {
	return "task.assignees.cleanup.team_removal"
}

// Handle cleans up task assignments and subscriptions for members removed from teams
func (l *CleanupTaskAssignmentsAfterTeamRemoval) Handle(msg *message.Message) (err error) {
	event := &TeamMemberRemovedEvent{}
	err = json.Unmarshal(msg.Payload, event)
	if err != nil {
		return err
	}

	s := db.NewSession()
	defer s.Close()

	if event == nil || event.Team == nil || event.Member == nil {
		return nil
	}

	err = cleanupTaskMembersAfterTeamRemoval(s, event.Team.ID, event.Member.ID)
	if err != nil {
		_ = s.Rollback()
		return err
	}

	return s.Commit()
}

// SendTeamMemberAddedNotification  represents a listener
type SendTeamMemberAddedNotification struct {
}

// Name defines the name for the SendTeamMemberAddedNotification listener
func (s *SendTeamMemberAddedNotification) Name() string {
	return "team.member.added.notification"
}

// Handle is executed when the event SendTeamMemberAddedNotification listens on is fired
func (s *SendTeamMemberAddedNotification) Handle(msg *message.Message) (err error) {
	event := &TeamMemberAddedEvent{}
	err = json.Unmarshal(msg.Payload, event)
	if err != nil {
		return err
	}

	// Don't notify the user themselves
	if event.Doer.ID == event.Member.ID {
		return nil
	}

	return notifications.Notify(event.Member, &TeamMemberAddedNotification{
		Member: event.Member,
		Doer:   event.Doer,
		Team:   event.Team,
	})
}

// HandleUserDataExport  represents a listener
type HandleUserDataExport struct {
}

// Name defines the name for the HandleUserDataExport listener
func (s *HandleUserDataExport) Name() string {
	return "handle.user.data.export"
}

// Handle is executed when the event HandleUserDataExport listens on is fired
func (s *HandleUserDataExport) Handle(msg *message.Message) (err error) {
	event := &UserDataExportRequestedEvent{}
	err = json.Unmarshal(msg.Payload, event)
	if err != nil {
		return err
	}

	log.Debugf("Starting to export user data for user %d...", event.User.ID)

	sess := db.NewSession()
	defer sess.Close()

	err = ExportUserData(sess, event.User)
	if err != nil {
		_ = sess.Rollback()
		return
	}

	log.Debugf("Done exporting user data for user %d...", event.User.ID)

	return sess.Commit()
}

type MarkTaskUnreadOnComment struct {
}

func (s *MarkTaskUnreadOnComment) Name() string {
	return "task.comment.mark.unread"
}

func (s *MarkTaskUnreadOnComment) Handle(msg *message.Message) (err error) {
	event := &TaskCommentCreatedEvent{}
	err = json.Unmarshal(msg.Payload, event)
	if err != nil {
		return err
	}

	sess := db.NewSession()
	defer sess.Close()

	project, err := GetProjectSimpleByID(sess, event.Task.ProjectID)
	if err != nil {
		_ = sess.Rollback()
		return err
	}

	users, err := ListUsersFromProject(sess, project, event.Doer, "")
	if err != nil {
		_ = sess.Rollback()
		return err
	}

	// Get existing unread statuses for this task
	existingUnreadStatuses := []*TaskUnreadStatus{}
	err = sess.
		Where("task_id = ?", event.Task.ID).
		Find(&existingUnreadStatuses)
	if err != nil {
		_ = sess.Rollback()
		return err
	}

	// Create a set of existing user IDs for quick lookup
	existingUserIDs := make(map[int64]bool)
	for _, status := range existingUnreadStatuses {
		existingUserIDs[status.UserID] = true
	}

	// Build list of new unread statuses
	unreadStatuses := []*TaskUnreadStatus{}
	for _, u := range users {
		// Skip the comment author and users who already have unread status
		if u.ID == event.Doer.ID || existingUserIDs[u.ID] {
			continue
		}
		unreadStatuses = append(unreadStatuses, &TaskUnreadStatus{
			TaskID: event.Task.ID,
			UserID: u.ID,
		})
	}

	// Bulk insert new unread statuses
	if len(unreadStatuses) > 0 {
		_, err = sess.Insert(&unreadStatuses)
		if err != nil {
			_ = sess.Rollback()
			return err
		}
	}

	return sess.Commit()
}
