package service import ( "context" "crypto/sha256" "database/sql" "encoding/hex" "encoding/json" "errors" "fmt" "os" "path/filepath" "regexp" "sort" "strings" "time" "internal-mail-skill/internal/db" "internal-mail-skill/internal/model" "internal-mail-skill/internal/parser" mailsmtp "internal-mail-skill/internal/smtp" ) type POP3Source interface { Fetch(ctx context.Context) ([]model.FetchedMessage, error) } type MailSender interface { Send(ctx context.Context, req mailsmtp.SendRequest) (mailsmtp.SendResult, error) } type Service struct { store *db.Store pop3 POP3Source mailer MailSender accountID string rawDir string attachmentDir string fromAddress string fromName string } type Options struct { RawDir string AttachmentDir string FromAddress string FromName string } func New(store *db.Store, pop3 POP3Source, mailer MailSender, accountID string) *Service { return NewWithOptions(store, pop3, mailer, accountID, Options{}) } func NewWithOptions(store *db.Store, pop3 POP3Source, mailer MailSender, accountID string, opts Options) *Service { if accountID == "" { accountID = "default" } return &Service{ store: store, pop3: pop3, mailer: mailer, accountID: accountID, rawDir: opts.RawDir, attachmentDir: opts.AttachmentDir, fromAddress: opts.FromAddress, fromName: opts.FromName, } } type SaveEmailInput struct { POP3UID string Raw []byte Parsed model.ParsedEmail } type SearchFilters struct { Q string From string To string After string Before string HasAttachment *bool Read *bool Processed *bool Archived *bool Replied *bool Limit int Offset int } type MarkInput struct { Read *bool `json:"read"` Processed *bool `json:"processed"` Archived *bool `json:"archived"` Deleted *bool `json:"deleted"` Replied *bool `json:"replied"` } type DraftInput struct { To []string `json:"to"` Cc []string `json:"cc"` Bcc []string `json:"bcc"` Subject string `json:"subject"` Body string `json:"body"` ReplyToEmailID *int64 `json:"reply_to_email_id,omitempty"` } type SyncResult struct { Total int `json:"total"` Stored int `json:"stored"` Skipped int `json:"skipped"` Errors []SyncError `json:"errors"` } type SyncError struct { UID string `json:"uid,omitempty"` ServerID int `json:"server_id,omitempty"` Error string `json:"error"` } type SendOutcome struct { SentMessage model.SentMessage `json:"sent_message"` Draft *model.Draft `json:"draft,omitempty"` } func (s *Service) Health(ctx context.Context) string { if err := s.store.Ping(ctx); err != nil { return "down" } return "ok" } func (s *Service) Sync(ctx context.Context) (SyncResult, error) { if s.pop3 == nil { return SyncResult{}, fmt.Errorf("POP3 client is not configured") } messages, err := s.pop3.Fetch(ctx) if err != nil { return SyncResult{}, err } result := SyncResult{Total: len(messages)} for _, fetched := range messages { if fetched.Error != "" { result.Errors = append(result.Errors, SyncError{UID: fetched.UID, ServerID: fetched.ServerID, Error: fetched.Error}) continue } exists, err := s.hasPOP3UID(ctx, fetched.UID) if err != nil { result.Errors = append(result.Errors, SyncError{UID: fetched.UID, ServerID: fetched.ServerID, Error: err.Error()}) continue } if exists { result.Skipped++ continue } parsed, err := parser.Parse(fetched.Raw) if err != nil { result.Errors = append(result.Errors, SyncError{UID: fetched.UID, ServerID: fetched.ServerID, Error: err.Error()}) continue } if _, err := s.SaveParsedEmail(ctx, SaveEmailInput{POP3UID: fetched.UID, Raw: fetched.Raw, Parsed: parsed}); err != nil { result.Errors = append(result.Errors, SyncError{UID: fetched.UID, ServerID: fetched.ServerID, Error: err.Error()}) continue } result.Stored++ } return result, nil } func (s *Service) SaveParsedEmail(ctx context.Context, input SaveEmailInput) (int64, error) { if input.POP3UID == "" { input.POP3UID = "local-" + hashBytes([]byte(input.Parsed.MessageIDHeader+input.Parsed.Subject+time.Now().String())) } if id, ok, err := s.findByPOP3UID(ctx, input.POP3UID); err != nil { return 0, err } else if ok { return id, nil } now := time.Now().UTC() rawPath := "" if len(input.Raw) > 0 && s.rawDir != "" { rawPath = filepath.Join(s.rawDir, s.accountID+"_"+hashBytes([]byte(input.POP3UID))+".eml") } toJSON, err := marshalJSON(input.Parsed.To) if err != nil { return 0, err } ccJSON, err := marshalJSON(input.Parsed.Cc) if err != nil { return 0, err } sentAt := timePtrString(input.Parsed.SentAt) tx, err := s.store.DB.BeginTx(ctx, nil) if err != nil { return 0, err } var writtenFiles []string committed := false defer func() { if !committed { _ = tx.Rollback() for _, path := range writtenFiles { _ = os.Remove(path) } } }() res, err := tx.ExecContext(ctx, ` INSERT INTO mail_messages ( account_id, pop3_uid, message_id_header, in_reply_to, references_header, subject, sender_name, sender_email, to_json, cc_json, date_header, sent_at, received_at, body_text, body_html, raw_eml_path, has_attachment, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, s.accountID, input.POP3UID, input.Parsed.MessageIDHeader, input.Parsed.InReplyTo, input.Parsed.References, input.Parsed.Subject, input.Parsed.SenderName, input.Parsed.SenderEmail, toJSON, ccJSON, input.Parsed.DateHeader, sentAt, formatTime(now), input.Parsed.BodyText, input.Parsed.BodyHTML, rawPath, boolInt(len(input.Parsed.Attachments) > 0), formatTime(now), formatTime(now), ) if err != nil { return 0, fmt.Errorf("insert mail message: %w", err) } id, err := res.LastInsertId() if err != nil { return 0, err } if rawPath != "" { if err := os.MkdirAll(s.rawDir, 0750); err != nil { return 0, fmt.Errorf("create raw mail dir: %w", err) } if err := os.WriteFile(rawPath, input.Raw, 0600); err != nil { return 0, fmt.Errorf("write raw eml: %w", err) } writtenFiles = append(writtenFiles, rawPath) } if err := s.saveAttachments(ctx, tx, id, input.Parsed.Attachments, now, &writtenFiles); err != nil { return 0, err } if err := tx.Commit(); err != nil { return 0, err } committed = true return id, nil } func (s *Service) SearchEmails(ctx context.Context, filters SearchFilters) ([]model.EmailSummary, error) { where := []string{"account_id = ?"} args := []any{s.accountID} if filters.Q != "" { where = append(where, "(subject LIKE ? OR sender_email LIKE ? OR body_text LIKE ? OR body_html LIKE ?)") like := "%" + filters.Q + "%" args = append(args, like, like, like, like) } if filters.From != "" { where = append(where, "(sender_email LIKE ? OR sender_name LIKE ?)") like := "%" + filters.From + "%" args = append(args, like, like) } if filters.To != "" { where = append(where, "to_json LIKE ?") args = append(args, "%"+filters.To+"%") } if after := normalizeDateFilter(filters.After, false); after != "" { where = append(where, "sent_at >= ?") args = append(args, after) } if before := normalizeDateFilter(filters.Before, true); before != "" { where = append(where, "sent_at <= ?") args = append(args, before) } addBoolFilter := func(name string, val *bool) { if val != nil { where = append(where, name+" = ?") args = append(args, boolInt(*val)) } } addBoolFilter("has_attachment", filters.HasAttachment) addBoolFilter("local_read", filters.Read) addBoolFilter("local_processed", filters.Processed) addBoolFilter("local_archived", filters.Archived) addBoolFilter("local_replied", filters.Replied) limit := filters.Limit if limit <= 0 || limit > 100 { limit = 50 } offset := filters.Offset if offset < 0 { offset = 0 } args = append(args, limit, offset) query := ` SELECT id, subject, sender_name, sender_email, to_json, cc_json, date_header, sent_at, received_at, body_text, has_attachment, local_read, local_processed, local_archived, local_deleted, local_replied, (SELECT COUNT(*) FROM attachments WHERE message_id = mail_messages.id) FROM mail_messages WHERE ` + strings.Join(where, " AND ") + ` ORDER BY COALESCE(sent_at, received_at) DESC, id DESC LIMIT ? OFFSET ?` rows, err := s.store.DB.QueryContext(ctx, query, args...) if err != nil { return nil, err } defer rows.Close() var out []model.EmailSummary for rows.Next() { summary, err := scanSummary(rows) if err != nil { return nil, err } out = append(out, summary) } if err := rows.Err(); err != nil { return nil, err } for i := range out { out[i].Labels = s.labelsForMessage(ctx, out[i].ID) } return out, nil } func (s *Service) GetEmail(ctx context.Context, id int64) (model.EmailDetail, error) { row := s.store.DB.QueryRowContext(ctx, ` SELECT id, message_id_header, in_reply_to, references_header, subject, sender_name, sender_email, to_json, cc_json, date_header, sent_at, received_at, body_text, body_html, local_read, local_processed, local_archived, local_deleted, local_replied FROM mail_messages WHERE account_id = ? AND id = ?`, s.accountID, id) detail, err := scanDetail(row) if err != nil { if errors.Is(err, sql.ErrNoRows) { return model.EmailDetail{}, ErrNotFound } return model.EmailDetail{}, err } detail.Attachments = s.attachmentsForMessage(ctx, id) detail.Labels = s.labelsForMessage(ctx, id) return detail, nil } func (s *Service) GetThread(ctx context.Context, id int64) ([]model.EmailSummary, error) { candidates, err := s.threadCandidates(ctx) if err != nil { return nil, err } if _, ok := candidates[id]; !ok { return nil, ErrNotFound } selected := map[int64]bool{id: true} changed := true for changed { changed = false for candidateID, candidate := range candidates { if selected[candidateID] { continue } for selectedID := range selected { if linked(candidate, candidates[selectedID]) { selected[candidateID] = true changed = true break } } } } out := make([]model.EmailSummary, 0, len(selected)) for selectedID := range selected { summary := candidates[selectedID].summary summary.Labels = s.labelsForMessage(ctx, selectedID) out = append(out, summary) } sort.Slice(out, func(i, j int) bool { ti := summaryTime(out[i]) tj := summaryTime(out[j]) if ti.Equal(tj) { return out[i].ID < out[j].ID } return ti.Before(tj) }) return out, nil } func (s *Service) MarkEmail(ctx context.Context, id int64, input MarkInput) error { sets := []string{} args := []any{} add := func(field string, value *bool) { if value != nil { sets = append(sets, field+" = ?") args = append(args, boolInt(*value)) } } add("local_read", input.Read) add("local_processed", input.Processed) add("local_archived", input.Archived) add("local_deleted", input.Deleted) add("local_replied", input.Replied) if len(sets) == 0 { return nil } sets = append(sets, "updated_at = ?") args = append(args, formatTime(time.Now().UTC()), s.accountID, id) res, err := s.store.DB.ExecContext(ctx, "UPDATE mail_messages SET "+strings.Join(sets, ", ")+" WHERE account_id = ? AND id = ?", args...) if err != nil { return err } return requireAffected(res) } func (s *Service) CreateLabel(ctx context.Context, name, color string) (model.Label, error) { name = strings.TrimSpace(name) if name == "" { return model.Label{}, fmt.Errorf("label name is required") } now := formatTime(time.Now().UTC()) _, err := s.store.DB.ExecContext(ctx, ` INSERT INTO labels(name, color, created_at, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(name) DO UPDATE SET color = excluded.color, updated_at = excluded.updated_at`, name, color, now, now) if err != nil { return model.Label{}, err } var label model.Label if err := s.store.DB.QueryRowContext(ctx, "SELECT id, name, color FROM labels WHERE name = ?", name).Scan(&label.ID, &label.Name, &label.Color); err != nil { return model.Label{}, err } return label, nil } func (s *Service) ApplyLabels(ctx context.Context, emailIDs, labelIDs []int64, op string) error { if len(emailIDs) == 0 || len(labelIDs) == 0 { return fmt.Errorf("email_ids and label_ids are required") } op = strings.ToLower(strings.TrimSpace(op)) if op != "add" && op != "remove" { return fmt.Errorf("op must be add or remove") } now := formatTime(time.Now().UTC()) for _, emailID := range emailIDs { for _, labelID := range labelIDs { if op == "add" { if _, err := s.store.DB.ExecContext(ctx, ` INSERT OR IGNORE INTO message_labels(message_id, label_id, created_at) VALUES (?, ?, ?)`, emailID, labelID, now); err != nil { return err } } else { if _, err := s.store.DB.ExecContext(ctx, `DELETE FROM message_labels WHERE message_id = ? AND label_id = ?`, emailID, labelID); err != nil { return err } } } } return nil } func (s *Service) CreateDraft(ctx context.Context, input DraftInput) (model.Draft, error) { now := formatTime(time.Now().UTC()) toJSON, _ := marshalJSON(input.To) ccJSON, _ := marshalJSON(input.Cc) bccJSON, _ := marshalJSON(input.Bcc) res, err := s.store.DB.ExecContext(ctx, ` INSERT INTO drafts(to_json, cc_json, bcc_json, subject, body, reply_to_email_id, status, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, 'draft', ?, ?)`, toJSON, ccJSON, bccJSON, input.Subject, input.Body, input.ReplyToEmailID, now, now) if err != nil { return model.Draft{}, err } id, err := res.LastInsertId() if err != nil { return model.Draft{}, err } return s.GetDraft(ctx, id) } func (s *Service) ListDrafts(ctx context.Context) ([]model.Draft, error) { rows, err := s.store.DB.QueryContext(ctx, ` SELECT id, to_json, cc_json, bcc_json, subject, body, reply_to_email_id, status, last_error, sent_message_id, created_at, updated_at FROM drafts ORDER BY updated_at DESC, id DESC`) if err != nil { return nil, err } defer rows.Close() var drafts []model.Draft for rows.Next() { draft, err := scanDraft(rows) if err != nil { return nil, err } drafts = append(drafts, draft) } return drafts, rows.Err() } func (s *Service) GetDraft(ctx context.Context, id int64) (model.Draft, error) { row := s.store.DB.QueryRowContext(ctx, ` SELECT id, to_json, cc_json, bcc_json, subject, body, reply_to_email_id, status, last_error, sent_message_id, created_at, updated_at FROM drafts WHERE id = ?`, id) draft, err := scanDraft(row) if err != nil { if errors.Is(err, sql.ErrNoRows) { return model.Draft{}, ErrNotFound } return model.Draft{}, err } return draft, nil } func (s *Service) UpdateDraft(ctx context.Context, id int64, input DraftInput) (model.Draft, error) { toJSON, _ := marshalJSON(input.To) ccJSON, _ := marshalJSON(input.Cc) bccJSON, _ := marshalJSON(input.Bcc) res, err := s.store.DB.ExecContext(ctx, ` UPDATE drafts SET to_json = ?, cc_json = ?, bcc_json = ?, subject = ?, body = ?, reply_to_email_id = ?, updated_at = ? WHERE id = ? AND status != 'sent'`, toJSON, ccJSON, bccJSON, input.Subject, input.Body, input.ReplyToEmailID, formatTime(time.Now().UTC()), id) if err != nil { return model.Draft{}, err } if err := requireAffected(res); err != nil { return model.Draft{}, err } return s.GetDraft(ctx, id) } func (s *Service) SendDraft(ctx context.Context, id int64) (SendOutcome, error) { draft, err := s.GetDraft(ctx, id) if err != nil { return SendOutcome{}, err } if draft.Status == "sent" { if draft.SentMessageID == nil { return SendOutcome{}, ErrDraftAlreadySent } sent, err := s.getSentMessage(ctx, *draft.SentMessageID) if err != nil { return SendOutcome{}, err } return SendOutcome{SentMessage: sent, Draft: &draft}, nil } req := mailsmtp.SendRequest{ FromAddress: s.fromAddress, FromName: s.fromName, To: draft.To, Cc: draft.Cc, Bcc: draft.Bcc, Subject: draft.Subject, Body: draft.Body, } if draft.ReplyToEmailID != nil { original, err := s.GetEmail(ctx, *draft.ReplyToEmailID) if err != nil { return SendOutcome{}, err } req.InReplyTo = original.MessageIDHeader req.References = buildReplyReferences(original) } var result mailsmtp.SendResult if s.mailer == nil { err = fmt.Errorf("SMTP client is not configured") } else { result, err = s.mailer.Send(ctx, req) } if err != nil { sent, recordErr := s.recordSentMessage(ctx, &id, draft, req, "failed", "", err.Error(), nil) if recordErr != nil { return SendOutcome{}, recordErr } _, _ = s.store.DB.ExecContext(ctx, "UPDATE drafts SET last_error = ?, updated_at = ? WHERE id = ?", err.Error(), formatTime(time.Now().UTC()), id) return SendOutcome{SentMessage: sent, Draft: &draft}, err } sentAt := result.SentAt sent, err := s.recordSentMessage(ctx, &id, draft, req, "sent", result.MessageIDHeader, "", &sentAt) if err != nil { return SendOutcome{}, err } _, err = s.store.DB.ExecContext(ctx, ` UPDATE drafts SET status = 'sent', last_error = NULL, sent_message_id = ?, updated_at = ? WHERE id = ?`, sent.ID, formatTime(time.Now().UTC()), id) if err != nil { return SendOutcome{}, err } if draft.ReplyToEmailID != nil { _ = s.MarkEmail(ctx, *draft.ReplyToEmailID, MarkInput{Replied: ptrBool(true)}) } updated, _ := s.GetDraft(ctx, id) return SendOutcome{SentMessage: sent, Draft: &updated}, nil } func (s *Service) SendDirect(ctx context.Context, input DraftInput) (SendOutcome, error) { draft, err := s.CreateDraft(ctx, input) if err != nil { return SendOutcome{}, err } return s.SendDraft(ctx, draft.ID) } type ForwardInput struct { To []string `json:"to"` Cc []string `json:"cc"` Bcc []string `json:"bcc"` Subject string `json:"subject"` Body string `json:"body"` IncludeAttachments bool `json:"include_attachments"` } func (s *Service) ForwardEmail(ctx context.Context, emailID int64, input ForwardInput) (SendOutcome, error) { if input.IncludeAttachments { return SendOutcome{}, ErrAttachmentsForwardUnsupported } original, err := s.GetEmail(ctx, emailID) if err != nil { return SendOutcome{}, err } subject := input.Subject if strings.TrimSpace(subject) == "" { subject = "Fwd: " + original.Subject } body := input.Body if strings.TrimSpace(body) != "" { body += "\n\n" } body += "---------- Forwarded message ----------\n" body += "From: " + original.SenderEmail + "\n" body += "Date: " + original.DateHeader + "\n" body += "Subject: " + original.Subject + "\n\n" if strings.TrimSpace(original.BodyText) != "" { body += original.BodyText } else { body += original.BodyHTML } return s.SendDirect(ctx, DraftInput{To: input.To, Cc: input.Cc, Bcc: input.Bcc, Subject: subject, Body: body}) } func (s *Service) hasPOP3UID(ctx context.Context, uid string) (bool, error) { _, ok, err := s.findByPOP3UID(ctx, uid) return ok, err } func (s *Service) findByPOP3UID(ctx context.Context, uid string) (int64, bool, error) { var id int64 err := s.store.DB.QueryRowContext(ctx, `SELECT id FROM mail_messages WHERE account_id = ? AND pop3_uid = ?`, s.accountID, uid).Scan(&id) if errors.Is(err, sql.ErrNoRows) { return 0, false, nil } if err != nil { return 0, false, err } return id, true, nil } func (s *Service) saveAttachments(ctx context.Context, execer interface { ExecContext(context.Context, string, ...any) (sql.Result, error) }, messageID int64, attachments []model.ParsedAttachment, now time.Time, writtenFiles *[]string) error { if len(attachments) == 0 { return nil } if s.attachmentDir == "" { for _, attachment := range attachments { if _, err := execer.ExecContext(ctx, ` INSERT INTO attachments(message_id, file_name, content_type, content_id, size_bytes, storage_path, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)`, messageID, attachment.Filename, attachment.ContentType, attachment.ContentID, attachment.SizeBytes, "", formatTime(now)); err != nil { return err } } return nil } dir := filepath.Join(s.attachmentDir, fmt.Sprintf("%d", messageID)) if err := os.MkdirAll(dir, 0750); err != nil { return fmt.Errorf("create attachment dir: %w", err) } for i, attachment := range attachments { filename := fmt.Sprintf("%03d_%s", i+1, attachment.Filename) path := filepath.Join(dir, filename) if err := os.WriteFile(path, attachment.Data, 0600); err != nil { return fmt.Errorf("write attachment: %w", err) } *writtenFiles = append(*writtenFiles, path) if _, err := execer.ExecContext(ctx, ` INSERT INTO attachments(message_id, file_name, content_type, content_id, size_bytes, storage_path, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)`, messageID, attachment.Filename, attachment.ContentType, attachment.ContentID, attachment.SizeBytes, path, formatTime(now)); err != nil { return err } } return nil } func (s *Service) recordSentMessage(ctx context.Context, draftID *int64, draft model.Draft, req mailsmtp.SendRequest, status, messageID, errorMessage string, sentAt *time.Time) (model.SentMessage, error) { toJSON, _ := marshalJSON(draft.To) ccJSON, _ := marshalJSON(draft.Cc) bccJSON, _ := marshalJSON(draft.Bcc) sentAtValue := timePtrString(sentAt) res, err := s.store.DB.ExecContext(ctx, ` INSERT INTO sent_messages(draft_id, to_json, cc_json, bcc_json, subject, body, message_id_header, in_reply_to, references_header, smtp_from, status, error_message, sent_at, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, draftID, toJSON, ccJSON, bccJSON, draft.Subject, draft.Body, messageID, req.InReplyTo, req.References, req.FromAddress, status, errorMessage, sentAtValue, formatTime(time.Now().UTC())) if err != nil { return model.SentMessage{}, err } id, err := res.LastInsertId() if err != nil { return model.SentMessage{}, err } return model.SentMessage{ ID: id, DraftID: draftID, Subject: draft.Subject, MessageIDHeader: messageID, Status: status, ErrorMessage: errorMessage, SentAt: sentAt, }, nil } func (s *Service) getSentMessage(ctx context.Context, id int64) (model.SentMessage, error) { var sent model.SentMessage var draftID sql.NullInt64 var sentAt, errorMessage sql.NullString err := s.store.DB.QueryRowContext(ctx, ` SELECT id, draft_id, subject, message_id_header, status, error_message, sent_at FROM sent_messages WHERE id = ?`, id).Scan( &sent.ID, &draftID, &sent.Subject, &sent.MessageIDHeader, &sent.Status, &errorMessage, &sentAt, ) if err != nil { if errors.Is(err, sql.ErrNoRows) { return model.SentMessage{}, ErrNotFound } return model.SentMessage{}, err } if draftID.Valid { sent.DraftID = &draftID.Int64 } if errorMessage.Valid { sent.ErrorMessage = errorMessage.String } if sentAt.Valid { sent.SentAt = parseStoredTime(sentAt.String) } return sent, nil } func scanSummary(rows interface{ Scan(dest ...any) error }) (model.EmailSummary, error) { var toJSON, ccJSON, sentAt, receivedAt, body string var hasAttachment, read, processed, archived, deleted, replied int var summary model.EmailSummary err := rows.Scan( &summary.ID, &summary.Subject, &summary.SenderName, &summary.SenderEmail, &toJSON, &ccJSON, &summary.DateHeader, &sentAt, &receivedAt, &body, &hasAttachment, &read, &processed, &archived, &deleted, &replied, &summary.AttachmentCount, ) if err != nil { return model.EmailSummary{}, err } summary.To = decodeAddresses(toJSON) summary.Cc = decodeAddresses(ccJSON) summary.SentAt = parseStoredTime(sentAt) summary.ReceivedAt = parseStoredTime(receivedAt) summary.Snippet = snippet(body) summary.HasAttachment = hasAttachment == 1 summary.LocalStatus = model.LocalStatus{Read: read == 1, Processed: processed == 1, Archived: archived == 1, Deleted: deleted == 1, Replied: replied == 1} return summary, nil } func scanDetail(row interface{ Scan(dest ...any) error }) (model.EmailDetail, error) { var toJSON, ccJSON, sentAt, receivedAt string var read, processed, archived, deleted, replied int var detail model.EmailDetail err := row.Scan( &detail.ID, &detail.MessageIDHeader, &detail.InReplyTo, &detail.References, &detail.Subject, &detail.SenderName, &detail.SenderEmail, &toJSON, &ccJSON, &detail.DateHeader, &sentAt, &receivedAt, &detail.BodyText, &detail.BodyHTML, &read, &processed, &archived, &deleted, &replied, ) if err != nil { return model.EmailDetail{}, err } detail.To = decodeAddresses(toJSON) detail.Cc = decodeAddresses(ccJSON) detail.SentAt = parseStoredTime(sentAt) detail.ReceivedAt = parseStoredTime(receivedAt) detail.LocalStatus = model.LocalStatus{Read: read == 1, Processed: processed == 1, Archived: archived == 1, Deleted: deleted == 1, Replied: replied == 1} return detail, nil } func scanDraft(row interface{ Scan(dest ...any) error }) (model.Draft, error) { var draft model.Draft var toJSON, ccJSON, bccJSON string var replyTo, sentMessage sql.NullInt64 var lastError sql.NullString var createdAt, updatedAt string err := row.Scan( &draft.ID, &toJSON, &ccJSON, &bccJSON, &draft.Subject, &draft.Body, &replyTo, &draft.Status, &lastError, &sentMessage, &createdAt, &updatedAt, ) if err != nil { return model.Draft{}, err } draft.To = decodeStrings(toJSON) draft.Cc = decodeStrings(ccJSON) draft.Bcc = decodeStrings(bccJSON) if replyTo.Valid { draft.ReplyToEmailID = &replyTo.Int64 } if sentMessage.Valid { draft.SentMessageID = &sentMessage.Int64 } if lastError.Valid { draft.LastError = lastError.String } draft.CreatedAt = parseStoredTime(createdAt) draft.UpdatedAt = parseStoredTime(updatedAt) return draft, nil } type threadCandidate struct { summary model.EmailSummary messageID string inReplyTo string references string } func (s *Service) threadCandidates(ctx context.Context) (map[int64]threadCandidate, error) { rows, err := s.store.DB.QueryContext(ctx, ` SELECT id, subject, sender_name, sender_email, to_json, cc_json, date_header, sent_at, received_at, body_text, has_attachment, local_read, local_processed, local_archived, local_deleted, local_replied, (SELECT COUNT(*) FROM attachments WHERE message_id = mail_messages.id) FROM mail_messages WHERE account_id = ?`, s.accountID) if err != nil { return nil, err } defer rows.Close() out := map[int64]threadCandidate{} for rows.Next() { summary, err := scanSummary(rows) if err != nil { return nil, err } out[summary.ID] = threadCandidate{summary: summary} } if err := rows.Err(); err != nil { return nil, err } refRows, err := s.store.DB.QueryContext(ctx, `SELECT id, message_id_header, in_reply_to, references_header FROM mail_messages WHERE account_id = ?`, s.accountID) if err != nil { return nil, err } defer refRows.Close() for refRows.Next() { var id int64 var messageID, inReplyTo, refs string if err := refRows.Scan(&id, &messageID, &inReplyTo, &refs); err != nil { return nil, err } c := out[id] c.messageID = messageID c.inReplyTo = inReplyTo c.references = refs out[id] = c } return out, refRows.Err() } func (s *Service) attachmentsForMessage(ctx context.Context, messageID int64) []model.Attachment { rows, err := s.store.DB.QueryContext(ctx, `SELECT id, message_id, file_name, content_type, content_id, size_bytes FROM attachments WHERE message_id = ? ORDER BY id`, messageID) if err != nil { return nil } defer rows.Close() var out []model.Attachment for rows.Next() { var item model.Attachment if err := rows.Scan(&item.ID, &item.MessageID, &item.Filename, &item.ContentType, &item.ContentID, &item.SizeBytes); err == nil { out = append(out, item) } } return out } func (s *Service) labelsForMessage(ctx context.Context, messageID int64) []model.Label { rows, err := s.store.DB.QueryContext(ctx, ` SELECT labels.id, labels.name, labels.color FROM labels JOIN message_labels ON message_labels.label_id = labels.id WHERE message_labels.message_id = ? ORDER BY labels.name`, messageID) if err != nil { return nil } defer rows.Close() var out []model.Label for rows.Next() { var label model.Label if err := rows.Scan(&label.ID, &label.Name, &label.Color); err == nil { out = append(out, label) } } return out } var messageIDPattern = regexp.MustCompile(`<[^>]+>`) func linked(a, b threadCandidate) bool { if a.messageID == "" || b.messageID == "" { return false } return refsContain(a.inReplyTo+" "+a.references, b.messageID) || refsContain(b.inReplyTo+" "+b.references, a.messageID) } func refsContain(refs, messageID string) bool { for _, ref := range extractMessageIDs(refs) { if ref == messageID { return true } } return false } func extractMessageIDs(value string) []string { matches := messageIDPattern.FindAllString(value, -1) if len(matches) > 0 { return matches } return strings.Fields(value) } func buildReplyReferences(original model.EmailDetail) string { refs := strings.TrimSpace(original.References) if original.MessageIDHeader == "" { return refs } if refs == "" { return original.MessageIDHeader } if refsContain(refs, original.MessageIDHeader) { return refs } return refs + " " + original.MessageIDHeader } func normalizeDateFilter(value string, endOfDay bool) string { value = strings.TrimSpace(value) if value == "" { return "" } if t, err := time.Parse(time.RFC3339Nano, value); err == nil { return formatTime(t.UTC()) } if t, err := time.Parse("2006-01-02", value); err == nil { if endOfDay { t = t.Add(24*time.Hour - time.Nanosecond) } return formatTime(t.UTC()) } return value } func marshalJSON(v any) (string, error) { data, err := json.Marshal(v) if err != nil { return "", err } return string(data), nil } func decodeAddresses(value string) []model.Address { var out []model.Address if err := json.Unmarshal([]byte(value), &out); err != nil { return nil } return out } func decodeStrings(value string) []string { var out []string if err := json.Unmarshal([]byte(value), &out); err != nil { return nil } return out } func boolInt(v bool) int { if v { return 1 } return 0 } func ptrBool(v bool) *bool { return &v } func formatTime(t time.Time) string { return t.UTC().Format(time.RFC3339Nano) } func timePtrString(t *time.Time) string { if t == nil { return "" } return formatTime(*t) } func parseStoredTime(value string) *time.Time { if value == "" { return nil } t, err := time.Parse(time.RFC3339Nano, value) if err != nil { return nil } return &t } func snippet(value string) string { value = strings.Join(strings.Fields(value), " ") if len(value) <= 160 { return value } return value[:160] } func hashBytes(data []byte) string { sum := sha256.Sum256(data) return hex.EncodeToString(sum[:16]) } func requireAffected(res sql.Result) error { affected, err := res.RowsAffected() if err != nil { return err } if affected == 0 { return ErrNotFound } return nil } func summaryTime(summary model.EmailSummary) time.Time { if summary.SentAt != nil { return *summary.SentAt } if summary.ReceivedAt != nil { return *summary.ReceivedAt } return time.Unix(0, summary.ID) } var ( ErrNotFound = errors.New("not found") ErrAttachmentsForwardUnsupported = errors.New("forwarding attachments is not supported in MVP") ErrDraftAlreadySent = errors.New("draft is already sent") )