1package imap
 2
 3import (
 4	"bytes"
 5	"context"
 6	"errors"
 7	"fmt"
 8	"io"
 9	"log/slog"
10	"net"
11	"time"
12
13	"postern/internal/db"
14	"postern/internal/model"
15	"postern/internal/policy"
16)
17
18func (i *Server) SubmitMessage(ctx context.Context, conn net.Conn, user model.User, r io.Reader) error {
19	logger := slog.With("component", "internal_submission", "user_id", user.ID)
20	internalTime := time.Now()
21
22	var msgBuffer bytes.Buffer
23	r = io.TeeReader(r, &msgBuffer)
24
25	fileName, size, msgHeader, err := i.persistence.WriteBlobMessage(r)
26	if err != nil {
27		return fmt.Errorf("write blob: %w", err)
28	}
29
30	policyUser := policy.UserToStarlark(user, i.db)
31	policyConn := policy.ConnToStarlark(conn)
32
33	msgBytes := msgBuffer.Bytes()
34	msg := &policy.Message{
35		OpenStream: func() (io.ReadCloser, error) {
36			return io.NopCloser(bytes.NewReader(msgBytes)), nil
37		},
38		Size: size,
39	}
40
41	destinationMailbox, err := i.policyEngine.OnMessageDeliver(policyConn, policyUser, msg)
42	if destinationMailbox == "" || err != nil {
43		logger.Warn("Policy script error, falling back to INBOX", "error", err)
44		destinationMailbox = "INBOX"
45	}
46
47	logger.InfoContext(ctx, "delivery", "mailbox", destinationMailbox)
48
49	// Create mailbox if not exists
50	mailboxID, _, err := i.db.GetMailboxID(ctx, user.ID, destinationMailbox)
51	if err != nil {
52		if errors.Is(err, db.ErrMailboxNotFound) {
53			err = i.db.CreateMailbox(ctx, user.ID, destinationMailbox)
54			if err != nil {
55				return fmt.Errorf("create on-demand mailbox: %w", err)
56			}
57			mailboxID, _, err = i.db.GetMailboxID(ctx, user.ID, destinationMailbox)
58			if err != nil {
59				return fmt.Errorf("get on-demand mailbox ID: %w", err)
60			}
61			err = i.db.Subscribe(ctx, user.ID, mailboxID, destinationMailbox)
62			if err != nil {
63				return fmt.Errorf("subscribe on-demand mailbox: %w", err)
64			}
65		} else {
66			return fmt.Errorf("get mailbox ID: %w", err)
67		}
68	}
69
70	_, err = i.db.AppendMessage(ctx, mailboxID, user.ID, fileName, size, internalTime.UTC().Format(time.RFC3339), nil, msgHeader)
71	if err != nil {
72		return fmt.Errorf("append message: %w", err)
73	}
74
75	// Notify user
76	t := i.mTracker.get(user.ID, mailboxID)
77	if t != nil {
78		mb, err := i.db.GetUserMailbox(ctx, user.ID, destinationMailbox)
79		if err != nil {
80			return fmt.Errorf("get user mailbox: %w", err)
81		}
82		t.QueueNumMessages(mb.NumMessages)
83		logger.Info("EXISTS message sent to user", "user", user.ID, "mailbox", mb.Name, "num_messages", mb.NumMessages)
84	}
85
86	return nil
87}