MailUI4Agents/gateway/internal/scheduler/calendar.go

package scheduler

import (
	"context"
	"errors"
	"fmt"
	"log"
	"strings"
	"sync"
	"time"

	"github.com/agentmail/gateway/internal/db"
	"github.com/agentmail/gateway/internal/models"
	"github.com/agentmail/gateway/internal/repo"
	"github.com/agentmail/gateway/internal/sse"
	"github.com/google/uuid"
)

// CalendarScheduler 定时扫描日历事件，触发到期提醒。
//
// 设计参照 Outlook 的 Exchange 提醒器：
//   - 每 30 秒扫描一次（精度到分钟够用，不需要秒级）
//   - 到期事件 → 生成一封提醒邮件 → 投递
//   - 重复事件自动推进到下一次
//   - 幂等：last_fired_at 保证同一分钟不触发两次
//
// 为什么用进程内 goroutine 而不是 cron：调度器与 Gateway 同生命周期，
// 不需要外部依赖，也不需要第二套「谁在跑」的信任模型。
var (
	schedMu   sync.Mutex
	schedStop chan struct{}
	schedWg   sync.WaitGroup
)

// Start 启动日历调度器。重复调用安全（先停旧的）。
func Start() {
	Stop()

	schedMu.Lock()
	defer schedMu.Unlock()

	schedStop = make(chan struct{})
	stop := schedStop
	schedWg.Add(1)

	go func() {
		defer schedWg.Done()
		runLoop(stop)
	}()

	log.Printf("[scheduler] 日历调度器已启动（30s 周期）")
}

// Stop 停止调度器，等待当前扫描完成。
func Stop() {
	schedMu.Lock()
	defer schedMu.Unlock()

	if schedStop != nil {
		close(schedStop)
		schedWg.Wait()
		schedStop = nil
		log.Printf("[scheduler] 日历调度器已停止")
	}
}

func runLoop(stop chan struct{}) {
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()

	// 启动时先扫一次：进程重启期间到期的事件不该被跳过
	scanAndFire()

	for {
		select {
		case <-stop:
			return
		case <-ticker.C:
			scanAndFire()
		}
	}
}

// scanAndFire 扫一轮到期事件。
//
// **全程不得 panic 出去**：它跑在后台 goroutine 里，而 Go 的 goroutine panic
// 会直接结束整个进程 —— 一条写坏的提醒把邮件网关整个带倒是荒谬的代价。
func scanAndFire() {
	defer func() {
		if rec := recover(); rec != nil {
			log.Printf("[scheduler] 扫描崩溃（已捕获，下一轮重试）: %v", rec)
		}
	}()

	// DB 未就绪就什么都不做。
	//
	// 调度器的启动时机比它看起来脆弱：main 里它排在 migrate 之后，
	// 但 Start() 是个导出函数，测试与将来的调用方都可能在建库前调到它。
	// 而 nil *sql.DB 上调 QueryContext 是空指针解引用，不是一个 error。
	if db.DB == nil {
		return
	}

	ctx := context.Background()
	events, err := repo.DueEvents(ctx)
	if err != nil {
		log.Printf("[scheduler] 查询到期事件失败: %v", err)
		return
	}
	for _, e := range events {
		fireEvent(ctx, e)
	}
}

// RenderReminder 把提醒模板里的变量替换成事件的实际内容。
//
// 单独提出来是为了可测：模板渲染错了不会立刻报错，
// 只会让 Agent 收到一封写着 `{title}` 的邮件。
//
// **{time} 必须转本地时区再格式化。**
// DSN 带 `_timezone=UTC`，从库里读回的 EventTime 是 UTC；直接 Format
// 会把人在 +0800 输入的 14:30 写成 06:30，而前端预览用的是本地时间
// —— 于是预览显示 14:30、Agent 收到 06:30，两边差 8 小时且两边都不报错。
// 日历是给人看的，人说「下午两点半」指的就是自己时区的那个时刻。
func RenderReminder(tmpl string, e models.CalendarEvent) string {
	body := tmpl
	body = strings.ReplaceAll(body, "{title}", e.Title)
	body = strings.ReplaceAll(body, "{time}", e.EventTime.Local().Format("2006-01-02 15:04"))
	body = strings.ReplaceAll(body, "{description}", e.Description)
	return body
}

// fireEvent 触发一条事件：渲染提醒文本、投递邮件、推进重复。
//
// 顺序很重要：**先标记已触发再发信**。
// 反过来的话，发信成功但标记失败会让下一轮再发一遍 ——
// 提醒邮件重复比漏发更糟（Agent 会把同一件事做两次）。
func fireEvent(ctx context.Context, e models.CalendarEvent) {
	// 单条事件崩溃不得让同一轮里剩下的提醒全部落空。
	defer func() {
		if rec := recover(); rec != nil {
			log.Printf("[scheduler] 事件 %s 触发崩溃（已捕获）: %v", short(e.EventID), rec)
		}
	}()

	// 收件人：Recipients 优先，为空时退回 to_address / agent_name（旧数据）
	recipients := e.EffectiveRecipients()
	if len(recipients) == 0 {
		log.Printf("[scheduler] 事件 %s 没有收件人，跳过（title=%q）", short(e.EventID), e.Title)
		// 没有收件人的事件永远发不出去，标记已触发免得每 30 秒重试一次
		_ = repo.MarkEventFired(ctx, e.EventID)
		return
	}

	if err := repo.MarkEventFired(ctx, e.EventID); err != nil {
		log.Printf("[scheduler] 标记事件 %s 已触发失败，本轮不发信: %v", short(e.EventID), err)
		return
	}

	body := RenderReminder(e.ReminderText, e)
	subject := "日程提醒：" + e.Title

	switch e.EffectiveDeliveryMode() {
	case models.DeliverTogether:
		// 一起发：首个是主收件人，其余进 cc_list —— 所有人共享同一条线索，
		// 能看到彼此的回复。适合「pi 主办、dsh 知情」这种有主次的协作。
		primary, cc := recipients[0], recipients[1:]
		if err := SendCalendarMail(ctx, e.EventID, primary, subject, body, e.CreatedBy, cc...); err != nil {
			log.Printf("[scheduler] 投递提醒失败（%s → %s +%d抄送）: %v",
				e.Title, primary, len(cc), err)
		} else {
			log.Printf("[scheduler] 已触发日历提醒: %s → %s（抄送 %d 人，同一线索）",
				e.Title, primary, len(cc))
		}

	default:
		// 各发一封：每人落在自己的会话里，互相看不到。
		// 适合「让三个 Agent 各自独立汇报」—— 用 together 会让他们互相
		// 看到回复而趋同，那种上下文污染事后无法分离。
		//
		// 一个失败不影响其余：三个 Agent 里有一个离线时，
		// 另外两个仍该收到提醒。
		ok, failed := 0, 0
		for _, addr := range recipients {
			if err := SendCalendarMail(ctx, e.EventID, addr, subject, body, e.CreatedBy); err != nil {
				failed++
				log.Printf("[scheduler] 投递提醒失败（%s → %s）: %v", e.Title, addr, err)
				continue
			}
			ok++
		}
		if failed == 0 {
			log.Printf("[scheduler] 已触发日历提醒: %s → %d 个收件人（各自独立会话）",
				e.Title, ok)
		} else {
			log.Printf("[scheduler] 日历提醒 %s：%d 成功 / %d 失败", e.Title, ok, failed)
		}
	}

	// 重复事件推进到下一次；一次性事件到此结束
	if e.Recurrence != models.RecurNone {
		advanced, err := repo.AdvanceRecurrence(ctx, e.EventID)
		if err != nil {
			log.Printf("[scheduler] 推进重复事件 %s 失败: %v", short(e.EventID), err)
		} else if !advanced {
			log.Printf("[scheduler] 重复事件 %s 已过 recurrence_end 或无法推进，已置为 cancelled",
				short(e.EventID))
		}
	}
}

func short(id string) string {
	if len(id) > 8 {
		return id[:8]
	}
	return id
}

// SendCalendarMail 把一条提醒投成邮件。
//
// 直接走 repo + sse 而不是自己发一个 HTTP 请求到 /mail/send：
// 调度器是进程内 goroutine，绕出去再进来只是多一次鉴权与序列化。
//
// **日历提醒不扣会话预算**：预算的语义是「这件事值得模型自主发多少封信」，
// 而提醒是人预先设定的定时任务，不是模型的自主行为。让它扣预算会出现
// 「每天 9 点的日报提醒把当天的预算吃掉一格」这种反直觉结果。
//
// 收件地址支持完整三维寻址：
//   - `agent`              → 该 Agent 的默认会话（长期提醒应该用这个，
//     所有提醒落在同一条线索上，模型看得到历史）
//   - `agent@/path`        → 指定工作目录的默认会话
//   - `agent@/path.alias`  → 指定已存在的会话（不存在则报错，不静默新建）
//   - `agent@/path.new`    → 每次提醒开一条新会话（适合互不相关的一次性任务）
func SendCalendarMail(ctx context.Context, eventID, toAddr, subject, body, createdBy string, ccAddrs ...string) error {
	addr, err := models.ParseAddress(toAddr)
	if err != nil {
		return fmt.Errorf("收件地址 %q 无法解析: %w", toAddr, err)
	}

	// 抄送方逐个解析。单个解析失败只跳过它，不让整封信发不出去 ——
	// 主收件人能收到提醒比「抄送名单必须完整」重要。
	ccList := make([]models.Address, 0, len(ccAddrs))
	for _, raw := range ccAddrs {
		raw = strings.TrimSpace(raw)
		if raw == "" {
			continue
		}
		ca, cErr := models.ParseAddress(raw)
		if cErr != nil {
			log.Printf("[scheduler] 抄送地址 %q 无法解析，已跳过: %v", raw, cErr)
			continue
		}
		// 抄送方的 session 位不参与会话定位（那是主收件人的事），
		// 但必须保留在地址里：cc_list 的 name/path 决定谁能看到这条线索。
		ccList = append(ccList, ca)
	}

	sessionID, err := resolveCalendarSession(ctx, addr, subject)
	if err != nil {
		return err
	}

	// 发件人固定为 calendar：它不是任何一个 Agent，也不是人。
	// 用创建者的名字会让 Agent 以为人在实时找它，而人此刻可能在睡觉 ——
	// 模型据此判断「要不要马上追问」，来源写错会让它问一个不在线的人。
	mailID, err := repo.CreateMail(ctx, sessionID, nil,
		calendarSender, "", addr.Name, addr.Path, subject, body, ccList)
	if err != nil {
		return err
	}

	// 事件附件复制成邮件附件。失败不阻断投递：提醒本身（正文）比附件重要得多，
	// 少一个附件也比整条提醒发不出去好 —— 后者会让人以为定时任务坏了。
	if eventID != "" {
		if n, aErr := repo.AttachCalendarFilesToMail(ctx, eventID, mailID, calendarSender); aErr != nil {
			log.Printf("[scheduler] 事件 %s 的附件挂载失败（提醒仍已发出）: %v", short(eventID), aErr)
		} else if n > 0 {
			log.Printf("[scheduler] 事件 %s 随提醒带了 %d 个附件", short(eventID), n)
		}
	}

	alias := repo.SessionAliasOf(ctx, sessionID)
	sse.Default.SendToRecipient(addr.Name, "new_mail", map[string]interface{}{
		"mail_id":       mailID.String(),
		"session_id":    sessionID.String(),
		"from_name":     calendarSender,
		"subject":       subject,
		"mail_type":     "normal",
		"role":          "to",
		"to_workspace":  addr.Path,
		"session_alias": alias,
		// 回信地址给 calendar 是发不出去的（它不是收件方），
		// 给会话自己的地址才让模型能把结果回报到同一条线索上。
		"reply_address": models.FormatAddress(addr.Name, addr.Path, alias),
		"self_address":  models.FormatAddress(addr.Name, addr.Path, alias),
		// 让插件与 UI 能区分「这封是定时提醒」而不是有人在找它
		"origin": "calendar",
	})
	sse.Default.SendToRecipient(addr.Name, "session_update", map[string]interface{}{
		"session_id": sessionID.String(),
		"status":     "active",
	})

	// 抄送方也要收到 SSE。
	//
	// 漏了这一步的后果很隐蔽：邮件的 cc_list 里有他们、他们**查**收件箱
	// 能看到这封信，但没有任何事件推给他们 —— 于是插件不会唤起会话，
	// Agent 直到下一次补拉（重启时）才发现。对「知情方」而言等于没通知。
	for _, c := range ccList {
		alias := repo.SessionAliasOf(ctx, sessionID)
		sse.Default.SendToRecipient(c.Name, "new_mail", map[string]interface{}{
			"mail_id":       mailID.String(),
			"session_id":    sessionID.String(),
			"from_name":     calendarSender,
			"subject":       subject,
			"mail_type":     "normal",
			"role":          "cc",
			"to_workspace":  c.Path,
			"session_alias": alias,
			"reply_address": models.FormatAddress(addr.Name, addr.Path, alias),
			"self_address":  models.FormatAddress(c.Name, c.Path, alias),
			"origin":        "calendar",
		})
		sse.Default.SendToRecipient(c.Name, "session_update", map[string]interface{}{
			"session_id": sessionID.String(),
			"status":     "active",
		})
	}

	// 创建者也该看到提醒发出去了 —— 否则「我设的提醒到底触发了没有」
	// 只能去翻 journalctl。
	if createdBy != "" && createdBy != addr.Name {
		sse.Default.SendToRecipient(createdBy, "session_update", map[string]interface{}{
			"session_id": sessionID.String(),
			"status":     "active",
		})
	}
	return nil
}

// calendarSender 是提醒邮件的发件人名。
//
// 刻意不是人类用户名也不是 Agent 名：这样收件方一眼能看出这封信来自定时任务，
// 而 `from_name` 又不会与命名空间里任何真实账号冲突。
const calendarSender = "calendar"

// resolveCalendarSession 按地址的 session 位定位会话。
//
// 与 handler.resolveTarget 同一套三态语义，但**不受新建会话速率限制**：
// 那条限制是防 Agent 暴开线索的，而日历事件的数量由人在界面上决定。
func resolveCalendarSession(ctx context.Context, addr models.Address, subject string) (uuid.UUID, error) {
	switch addr.Mode() {
	case models.SessionNew:
		id, err := repo.CreateSession(ctx, nil, calendarSender, subject, addr.Path)
		if err != nil {
			return uuid.Nil, err
		}
		// 与发信路径一致：`.new` 建完必须立刻有别名，否则这条会话
		// 除了回复那一封之外再也无法寻址（未命名会话查不到也补全不出来）。
		_, _ = repo.EnsureSessionAlias(ctx, id, repo.AutoAliasFor(addr.Name, subject))
		return id, nil

	case models.SessionNamed:
		id, err := repo.FindNamedSessionFor(ctx, addr.Name, addr.Path, addr.Session)
		if errors.Is(err, repo.ErrSessionNotFound) {
			// 不静默新建：别名指向不存在的会话时报错，与人发信时的语义一致。
			// 静默新建会让「提醒发到哪去了」变成一个查不清的问题。
			return uuid.Nil, fmt.Errorf("会话 %q 不存在于 %s@%s", addr.Session, addr.Name, addr.Path)
		}
		if err != nil {
			return uuid.Nil, err
		}
		repo.TouchSession(ctx, id)
		return id, nil

	default: // SessionDefault
		id, err := repo.FindOrCreateDefaultSession(ctx, addr.Name, addr.Path, calendarSender, subject)
		if err != nil {
			return uuid.Nil, err
		}
		_, _ = repo.EnsureSessionAlias(ctx, id, repo.AutoAliasFor(addr.Name, subject))
		return id, nil
	}
}