From 5ccca2fac90c4ec70c624a8748816452578f47f9 Mon Sep 17 00:00:00 2001 From: Andriy Oblivantsev Date: Fri, 14 Aug 2026 16:35:20 +0000 Subject: [PATCH] feat(mail): add Microsoft 365 Graph sync source with delta state Adds a GET-only Microsoft Graph source to bin/mail/sync (client-credentials, delta query). Wires it into the CLI (--source m365), the worker pool and the compose stack: - m365.go: M365Client (token cache, delta pagination, message normalization, attachment download) + m365Source adapter (per-mailbox folders + delta link) - sync.go: M365Credentials in SyncConfig; Committer interface so delta links only advance after a fully successful run (no skips on failure) - cli.go: --source m365 with M365_/MS_ env passthrough - Dockerfile: mail-build stage produces /mail-sync into the index image - docker-entrypoint: mail-sync loop (sync -> import -> index, default 10s) - compose.yaml: mail-sync service (index image, kb-var volume, secrets ro) --- .gitignore | 6 + Dockerfile | 9 + bin/docker-entrypoint | 20 ++ bin/mail/sync.go | 5 +- bin/mail/sync/cli.go | 25 ++- bin/mail/sync/m365.go | 382 +++++++++++++++++++++++++++++++++++++ bin/mail/sync/m365_test.go | 188 ++++++++++++++++++ bin/mail/sync/sync.go | 44 ++++- compose.yaml | 27 +++ 9 files changed, 700 insertions(+), 6 deletions(-) create mode 100644 bin/mail/sync/m365.go create mode 100644 bin/mail/sync/m365_test.go diff --git a/.gitignore b/.gitignore index 001fb10..5cc73a0 100644 --- a/.gitignore +++ b/.gitignore @@ -14,3 +14,9 @@ go.work.local models/ # Purged from git history. Do not re-add. docs/crm-associations-proof.md + +# mount scaffold for the 8TB volume, never part of the repo +mnt/ + +# go build ./bin/mail/sync.go drops a binary named `sync` in cwd +/sync diff --git a/Dockerfile b/Dockerfile index 61366a6..20cc43b 100644 --- a/Dockerfile +++ b/Dockerfile @@ -28,6 +28,7 @@ RUN python -m pip install --no-cache-dir -r /tmp/requirements.lock.txt \ COPY . . RUN chmod +x /app/bin/docker-entrypoint \ && chown -R 2dph:2dph /app +COPY --from=mail-build /mail-sync /app/bin/mail-sync USER 2dph ENV PATH="/app/bin:${PATH}" \ @@ -37,6 +38,14 @@ HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ CMD python -c "import model2vec, ladybug, mistune; print('ok')" || exit 1 ENTRYPOINT ["/app/bin/docker-entrypoint"] +# --- mail-sync: standalone M365/OnlyOffice/Gmail puller (pure Go, no CGO) --- +FROM golang:1.26-bookworm AS mail-build +WORKDIR /src +COPY go.mod go.sum ./ +RUN go mod download +COPY bin/mail ./bin/mail +RUN CGO_ENABLED=0 go build -o /mail-sync ./bin/mail/sync.go + # --- Go API: CGO with Zig, not gcc --- FROM golang:1.26-bookworm AS api-build WORKDIR /src diff --git a/bin/docker-entrypoint b/bin/docker-entrypoint index a7e3256..9cdacb4 100755 --- a/bin/docker-entrypoint +++ b/bin/docker-entrypoint @@ -5,6 +5,7 @@ # serve | search | watch # Index image (Python write path, compose profile `index`): # index | extract | audit | search (deprecated python wrapper) +# mail-sync [N] pull loop: sync -> import -> index every N s (default 10) # # Usage comment starts at line 2 (self-describing convention). set -euo pipefail @@ -34,5 +35,24 @@ case "$CMD" in serve) exec /app/bin/serve "$@" ;; extract) exec "$KB_PY" /app/bin/facts/extract "$@" ;; audit) exec "$KB_PY" /app/bin/facts/audit "$@" ;; + mail-sync) + # loop: pull mail (M365/OnlyOffice/Gmail) every N s, convert to md, + # rebuild the brain index. Index only when something new arrived. + interval="${1:-10}" + [ "$interval" -gt 0 ] 2>/dev/null || interval=10 + : "${MAIL_SYNC_ENV:=/secret/m365.env}" + : "${MAIL_SYNC_SRC:=m365}" + : "${MAIL_SYNC_OUT:=/app/var/mail}" + while true; do + out="$("/app/bin/mail-sync" --source "$MAIL_SYNC_SRC" --env "$MAIL_SYNC_ENV" --out "$MAIL_SYNC_OUT" 2>&1)" + echo "$out" + new="$(printf '%s\n' "$out" | sed -n 's/.*new=\([0-9]*\).*/\1/p' | tail -1)" + if [ -n "$new" ] && [ "$new" -gt 0 ] 2>/dev/null; then + "$KB_PY" /app/bin/mail/import --from-raw "$MAIL_SYNC_OUT" 2>&1 | tail -1 + "$KB_PY" /app/bin/kb/index --rebuild --with-mail 2>&1 | tail -1 + fi + sleep "$interval" + done + ;; *) echo "unknown command: $CMD" >&2; exit 2 ;; esac diff --git a/bin/mail/sync.go b/bin/mail/sync.go index f03d989..82aa64e 100755 --- a/bin/mail/sync.go +++ b/bin/mail/sync.go @@ -1,8 +1,9 @@ //usr/bin/env go run "$0" "$@"; exit -// bin/mail/sync.go - async download of OnlyOffice and Gmail mail to var/mail/. +// bin/mail/sync.go - async download of OnlyOffice, Gmail and M365 mail to var/mail/. // -// ./bin/mail/sync.go --source onlyoffice,gmail --limit 50 --workers 8 +// ./bin/mail/sync.go --source onlyoffice,gmail,m365 --limit 50 --workers 8 // ./bin/mail/sync.go --source gmail --force +// ./bin/mail/sync.go --source m365 --env .secrets/m365.env // ./bin/mail/sync.go --dry-run // // Writes raw message.json + attachments under var/mail///; run diff --git a/bin/mail/sync/cli.go b/bin/mail/sync/cli.go index 9ec41ec..eec710d 100644 --- a/bin/mail/sync/cli.go +++ b/bin/mail/sync/cli.go @@ -55,7 +55,7 @@ func bind(v *flagVals) *flaggy.Parser { p.Bool(&v.force, "", "force", "overwrite existing message.json") p.Bool(&v.dryRun, "", "dry-run", "list counts without writing") p.String(&v.query, "", "query", "Gmail search query") - p.String(&v.srcs, "", "source", "comma list: onlyoffice,gmail") + p.String(&v.srcs, "", "source", "comma list: onlyoffice,gmail,m365") return p } @@ -110,6 +110,24 @@ func ParseCLI(args []string) (CLIConfig, int, error) { CredentialsPath: filepath.Join(home, ".gmail-mcp", "credentials.json"), KeysPath: filepath.Join(home, ".gmail-mcp", "gcp-oauth.keys.json"), } + case "m365": + tenant := pick(envVars["M365_TENANT"], envVars["MS_TENANT"]) + cid := pick(envVars["M365_CLIENT_ID"], envVars["MS_CLIENT_ID"]) + sec := pick(envVars["M365_CLIENT_SECRET"], envVars["MS_CLIENT_SECRET"]) + users := pick(envVars["M365_USERS"], envVars["MS_USERS"]) + if tenant == "" || cid == "" || sec == "" || users == "" { + return CLIConfig{}, 2, fmt.Errorf("m365 source needs M365_TENANT/CLIENT_ID/CLIENT_SECRET/USERS in %s", v.env) + } + var userList []string + for _, u := range strings.Split(users, ",") { + if u = strings.TrimSpace(u); u != "" { + userList = append(userList, u) + } + } + if len(userList) == 0 { + return CLIConfig{}, 2, fmt.Errorf("m365 source: M365_USERS empty") + } + cfg.M365 = &M365Credentials{Tenant: tenant, ClientID: cid, ClientSecret: sec, Users: userList} default: return CLIConfig{}, 2, fmt.Errorf("unknown source %q", s) } @@ -126,7 +144,7 @@ func Main(args []string) int { return code } if cfg.Help { - fmt.Fprintln(os.Stderr, "usage: bin/mail/sync.go [--source onlyoffice,gmail] [--query GMAIL_Q] [--limit N] [--offset N] [--workers N] [--force] [--dry-run]") + fmt.Fprintln(os.Stderr, "usage: bin/mail/sync.go [--source onlyoffice,gmail,m365] [--query GMAIL_Q] [--limit N] [--offset N] [--workers N] [--force] [--dry-run]") return 0 } ctx, cancel := context.WithTimeout(context.Background(), 6*time.Hour) @@ -169,7 +187,8 @@ func readEnv(path string) map[string]string { if !ok { continue } - if strings.HasPrefix(k, "ONLYOFFICE_") || strings.HasPrefix(k, "OO_") { + if strings.HasPrefix(k, "ONLYOFFICE_") || strings.HasPrefix(k, "OO_") || + strings.HasPrefix(k, "M365_") || strings.HasPrefix(k, "MS_") { out[k] = v } } diff --git a/bin/mail/sync/m365.go b/bin/mail/sync/m365.go new file mode 100644 index 0000000..f3add04 --- /dev/null +++ b/bin/mail/sync/m365.go @@ -0,0 +1,382 @@ +package sync + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "os" + "path/filepath" + "strings" + "time" +) + +// M365Credentials holds a Microsoft Graph app registration with the Mail.Read +// application permission. Client credentials are read from env/.env, never +// committed. +type M365Credentials struct { + Tenant string + ClientID string + ClientSecret string + Users []string // mailbox addresses to sync, e.g. info@example.com +} + +// m365Token is the cached access token with expiry. +type m365Token struct { + AccessToken string + Expiry time.Time +} + +// M365Client talks to the Microsoft Graph API using the client-credentials +// flow (app registration with Mail.Read application permission). GET-only: +// messages are never marked as read or deleted. +type M365Client struct { + creds M365Credentials + base string // graph base URL; default https://graph.microsoft.com + tokenEndpoint string // login endpoint; default https://login.microsoftonline.com + client *http.Client + mu chan struct{} + token *m365Token +} + +func NewM365Client(creds M365Credentials) (*M365Client, error) { + if creds.Tenant == "" || creds.ClientID == "" || creds.ClientSecret == "" { + return nil, errors.New("m365 needs tenant, client id and client secret") + } + c := &M365Client{ + creds: creds, + base: "https://graph.microsoft.com", + client: &http.Client{Timeout: 90 * time.Second}, + mu: make(chan struct{}, 1), + } + c.mu <- struct{}{} + return c, nil +} + +// accessToken returns a fresh bearer token, refreshing via the Azure AD token +// endpoint when the cached one is missing or about to expire (within 2 min). +func (c *M365Client) accessToken(ctx context.Context) (string, error) { + select { + case <-c.mu: + case <-ctx.Done(): + return "", ctx.Err() + } + defer func() { c.mu <- struct{}{} }() + if c.token != nil && c.token.AccessToken != "" && time.Now().Before(c.token.Expiry.Add(-2*time.Minute)) { + return c.token.AccessToken, nil + } + return c.refreshLocked(ctx) +} + +func (c *M365Client) refreshLocked(ctx context.Context) (string, error) { + form := url.Values{} + form.Set("grant_type", "client_credentials") + form.Set("client_id", c.creds.ClientID) + form.Set("client_secret", c.creds.ClientSecret) + form.Set("scope", "https://graph.microsoft.com/.default") + + endpoint := c.tokenEndpoint + if endpoint == "" { + endpoint = fmt.Sprintf("https://login.microsoftonline.com/%s/oauth2/v2.0/token", c.creds.Tenant) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, strings.NewReader(form.Encode())) + if err != nil { + return "", err + } + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + resp, err := c.client.Do(req) + if err != nil { + return "", fmt.Errorf("m365 token: %w", err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if resp.StatusCode != http.StatusOK { + var e struct { + Error string `json:"error"` + Desc string `json:"error_description"` + } + _ = json.Unmarshal(body, &e) + return "", fmt.Errorf("m365 token status %d: %s (%s)", resp.StatusCode, e.Error, truncate(e.Desc, 200)) + } + var out struct { + AccessToken string `json:"access_token"` + ExpiresIn int64 `json:"expires_in"` + } + if err := json.Unmarshal(body, &out); err != nil { + return "", fmt.Errorf("m365 token parse: %w", err) + } + c.token = &m365Token{ + AccessToken: out.AccessToken, + Expiry: time.Now().Add(time.Duration(out.ExpiresIn) * time.Second), + } + return out.AccessToken, nil +} + +// deltaPage is one response page of the Graph delta query. +type deltaPage struct { + Value []struct { + ID string `json:"id"` + RemovedReason string `json:"@odata.removedReason"` + } `json:"value"` + NextLink string `json:"@odata.nextLink"` + DeltaLink string `json:"@odata.deltaLink"` +} + +// ListDeltaIDs walks the inbox delta query and returns live message ids since +// the previous deltaLink (or the full inbox when deltaLink is empty). Returns +// the new deltaLink for the next run. GET-only; nothing is mutated server-side. +func (c *M365Client) ListDeltaIDs(ctx context.Context, mailbox, deltaLink string, limit int) ([]string, string, error) { + var ( + ids []string + url string + link = deltaLink + ) + if link == "" { + url = fmt.Sprintf("/v1.0/users/%s/mailFolders/inbox/messages/delta", pathEscape(mailbox)) + } else { + url = link + } + for url != "" { + var page deltaPage + if err := c.getJSON(ctx, url, &page); err != nil { + return ids, link, err + } + for _, m := range page.Value { + if m.RemovedReason != "" { + continue + } + if m.ID == "" { + continue + } + ids = append(ids, m.ID) + if limit > 0 && len(ids) >= limit { + if page.DeltaLink != "" { + link = page.DeltaLink + } + return ids, link, nil + } + } + if page.DeltaLink != "" { + link = page.DeltaLink + url = "" + break + } + url = page.NextLink + } + return ids, link, nil +} + +// GetMessage fetches a single message by id and normalizes it to the Message +// contract. GET-only. +func (c *M365Client) GetMessage(ctx context.Context, mailbox, id string) (*Message, error) { + path := fmt.Sprintf("/v1.0/users/%s/messages/%s?$expand=attachments($select=id,name,contentType,size,isInline)", + pathEscape(mailbox), pathEscape(id)) + var raw struct { + ID string `json:"id"` + Subject string `json:"subject"` + From m365Recipient `json:"from"` + ToRecipients []m365Recipient `json:"toRecipients"` + CCRecipients []m365Recipient `json:"ccRecipients"` + BCCRecipients []m365Recipient `json:"bccRecipients"` + ReceivedDateTime string `json:"receivedDateTime"` + Body m365Body `json:"body"` + BodyPreview string `json:"bodyPreview"` + InternetMessageID string `json:"internetMessageId"` + Attachments []m365Attachment `json:"attachments"` + } + if err := c.getJSON(ctx, path, &raw); err != nil { + return nil, err + } + m := &Message{ + Source: "m365", + ID: raw.ID, + Folder: "m365", + Subject: raw.Subject, + From: formatRecipient(raw.From), + To: formatRecipients(raw.ToRecipients), + CC: formatRecipients(raw.CCRecipients), + BCC: formatRecipients(raw.BCCRecipients), + MimeMessageID: raw.InternetMessageID, + } + if t, err := time.Parse(time.RFC3339, raw.ReceivedDateTime); err == nil { + m.ReceivedAt = t + } + switch strings.ToLower(raw.Body.ContentType) { + case "html": + m.HTMLBody = raw.Body.Content + if raw.BodyPreview != "" { + m.TextBody = raw.BodyPreview + } + default: + m.TextBody = raw.Body.Content + if raw.BodyPreview != "" { + m.HTMLBody = raw.BodyPreview + } + } + for _, a := range raw.Attachments { + if a.IsInline || a.ID == "" || a.Name == "" { + continue + } + m.Attachments = append(m.Attachments, Attachment{ + FileID: a.ID, + FileName: a.Name, + StoredName: a.Name, + Size: a.Size, + ContentType: a.ContentType, + }) + } + m.HasAttachments = len(m.Attachments) > 0 + return m, nil +} + +type m365Recipient struct { + EmailAddress struct { + Name string `json:"name"` + Address string `json:"address"` + } `json:"emailAddress"` +} + +type m365Body struct { + ContentType string `json:"contentType"` + Content string `json:"content"` +} + +type m365Attachment struct { + ID string `json:"id"` + Name string `json:"name"` + ContentType string `json:"contentType"` + Size int64 `json:"size"` + IsInline bool `json:"isInline"` + ContentID string `json:"contentId"` +} + +func formatRecipients(rs []m365Recipient) string { + var parts []string + for _, r := range rs { + if s := formatRecipient(r); s != "" { + parts = append(parts, s) + } + } + return strings.Join(parts, ", ") +} + +func formatRecipient(r m365Recipient) string { a := r.EmailAddress.Address + n := r.EmailAddress.Name + switch { + case n == "" || n == a: + return a + case a == "": + return n + default: + return fmt.Sprintf("%s <%s>", n, a) + } +} + +// DownloadAttachment fetches an attachment's raw bytes via the /$value stream. +func (c *M365Client) DownloadAttachment(ctx context.Context, mailbox, msgID, attID string) ([]byte, error) { + path := fmt.Sprintf("/v1.0/users/%s/messages/%s/attachments/%s/$value", + pathEscape(mailbox), pathEscape(msgID), pathEscape(attID)) + tok, err := c.accessToken(ctx) + if err != nil { + return nil, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.base+path, nil) + if err != nil { + return nil, err + } + req.Header.Set("Authorization", "Bearer "+tok) + resp, err := c.client.Do(req) + if err != nil { + return nil, fmt.Errorf("m365 attachment: %w", err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(io.LimitReader(resp.Body, 256<<20)) + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("m365 attachment %s: status %d: %s", attID, resp.StatusCode, truncate(string(body), 300)) + } + return body, nil +} + +func (c *M365Client) getJSON(ctx context.Context, path string, out any) error { + tok, err := c.accessToken(ctx) + if err != nil { + return err + } + u := path + if !strings.HasPrefix(u, "http") { + u = c.base + u + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+tok) + resp, err := c.client.Do(req) + if err != nil { + return fmt.Errorf("m365 %s: %w", path, err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(io.LimitReader(resp.Body, 16<<20)) + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("m365 %s: status %d: %s", path, resp.StatusCode, truncate(string(body), 300)) + } + if out != nil { + return json.Unmarshal(body, out) + } + return nil +} + +// m365Source adapts a mailbox to the Source worker-pool contract. Each mailbox +// gets its own folder under var/mail/m365// and a delta state file. +type m365Source struct { + c *M365Client + mailbox string + localpart string + stateDir string + pending string // delta link to persist on Commit() + hasPending bool +} + +func (s *m365Source) Folder() string { return filepath.Join("m365", s.localpart) } + +func (s *m365Source) ListIDs(ctx context.Context, limit int, cursor string) ([]string, string, error) { + link, _ := os.ReadFile(filepath.Join(s.stateDir, s.localpart+".deltalink")) + ids, newLink, err := s.c.ListDeltaIDs(ctx, s.mailbox, strings.TrimSpace(string(link)), limit) + if err != nil { + return nil, "", err + } + // Buffer the new delta link; persist it only in Commit() after the full + // batch downloaded, so a failed run stays retryable without gaps. + if newLink != "" { + s.pending = newLink + s.hasPending = true + } + return ids, "", nil +} + +// Commit persists the buffered delta link. Called by the sync runner only when +// every listed message downloaded successfully. +func (s *m365Source) Commit() error { + if !s.hasPending || s.pending == "" { + return nil + } + if err := os.MkdirAll(s.stateDir, 0o755); err != nil { + return err + } + return os.WriteFile(filepath.Join(s.stateDir, s.localpart+".deltalink"), []byte(s.pending), 0o644) +} + +func (s *m365Source) Get(ctx context.Context, id string) (*Message, error) { + return s.c.GetMessage(ctx, s.mailbox, id) +} + +func (s *m365Source) DownloadAttachment(ctx context.Context, msg *Message, att Attachment) ([]byte, error) { + return s.c.DownloadAttachment(ctx, s.mailbox, msg.ID, att.FileID) +} + +func pathEscape(s string) string { + return url.PathEscape(s) +} diff --git a/bin/mail/sync/m365_test.go b/bin/mail/sync/m365_test.go new file mode 100644 index 0000000..9e52686 --- /dev/null +++ b/bin/mail/sync/m365_test.go @@ -0,0 +1,188 @@ +package sync + +import ( + "context" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" +) + +// newM365TestClient serves the Graph delta + message endpoints against a fake +// token endpoint, so unit tests never touch the network. +func newM365TestClient(t *testing.T, graph http.Handler) *M365Client { + t.Helper() + tok := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"access_token":"test-token","expires_in":3600}`)) + })) + gr := httptest.NewServer(graph) + t.Cleanup(func() { + tok.Close() + gr.Close() + }) + client, err := NewM365Client(M365Credentials{Tenant: "t.onmicrosoft.com", ClientID: "c", ClientSecret: "s"}) + if err != nil { + t.Fatal(err) + } + client.base = gr.URL + client.tokenEndpoint = tok.URL + return client +} + +func TestM365AccessToken(t *testing.T) { + c := newM365TestClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + got, err := c.accessToken(context.Background()) + if err != nil { + t.Fatalf("accessToken: %v", err) + } + if got != "test-token" { + t.Errorf("token = %q", got) + } + // Second call must reuse the cached token (no token request). + again, err := c.accessToken(context.Background()) + if err != nil || again != "test-token" { + t.Fatalf("cached token: %q, %v", again, err) + } +} + +func TestM365DeltaSkipsTombstones(t *testing.T) { + first := true + graph := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + if first { + first = false + w.Write([]byte(`{"value":[ + {"id":"m1"}, + {"id":"m2","@odata.removedReason":"deleted"}, + {"id":"m3"} + ],"@odata.deltaLink":"` + deltaNext + `"}`)) + return + } + // Second call must use the stored deltaLink (points at this server). + if r.URL.Path != "/v1.0/delta-next" { + w.WriteHeader(500) + w.Write([]byte(`{"error":{"message":"unexpected path"}}`)) + return + } + w.Write([]byte(`{"value":[{"id":"m4"}],"@odata.deltaLink":"` + deltaFinal + `"}`)) + }) + c := newM365TestClient(t, graph) + deltaNext = c.base + "/v1.0/delta-next" + deltaFinal = c.base + "/v1.0/delta-final" + + ids, link, err := c.ListDeltaIDs(context.Background(), "a@x.de", "", 0) + if err != nil { + t.Fatalf("delta: %v", err) + } + if len(ids) != 2 || ids[0] != "m1" || ids[1] != "m3" { + t.Errorf("ids = %v", ids) + } + if link == "" { + t.Error("expected new deltaLink") + } + // Incremental: pass the deltaLink, get only the new id. + ids2, link2, err := c.ListDeltaIDs(context.Background(), "a@x.de", link, 0) + if err != nil { + t.Fatalf("delta incremental: %v", err) + } + if len(ids2) != 1 || ids2[0] != "m4" { + t.Errorf("ids2 = %v", ids2) + } + if link2 == "" { + t.Error("expected updated deltaLink") + } +} + +func TestM365GetMessageNormalizes(t *testing.T) { + graph := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + if pathLast(r.URL.Path) == "messages" { + w.Write([]byte(`{"value":[{"id":"m1"}]}`)) + return + } + w.Write([]byte(`{ + "id":"m1", + "subject":"Hallo", + "from":{"emailAddress":{"name":"Max","address":"max@x.de"}}, + "toRecipients":[{"emailAddress":{"address":"a@x.de"}}], + "receivedDateTime":"2026-08-14T08:15:00Z", + "body":{"contentType":"html","content":"

body

"}, + "bodyPreview":"body", + "internetMessageId":"", + "attachments":[ + {"id":"att1","name":"doc.pdf","contentType":"application/pdf","size":10}, + {"id":"img1","name":"logo.png","contentType":"image/png","isInline":true} + ] + }`)) + }) + c := newM365TestClient(t, graph) + m, err := c.GetMessage(context.Background(), "a@x.de", "m1") + if err != nil { + t.Fatalf("GetMessage: %v", err) + } + if m.ID != "m1" || m.Subject != "Hallo" || m.From != "Max " || m.To != "a@x.de" { + t.Errorf("headers mismatch: %+v", m) + } + if m.HTMLBody != "

body

" { + t.Errorf("html = %q", m.HTMLBody) + } + if m.ReceivedAt.IsZero() { + t.Error("receivedAt zero") + } + if m.MimeMessageID != "" { + t.Errorf("mime id = %q", m.MimeMessageID) + } + if len(m.Attachments) != 1 || m.Attachments[0].FileName != "doc.pdf" || m.Attachments[0].FileID != "att1" { + t.Errorf("atts = %+v", m.Attachments) + } + if !m.HasAttachments { + t.Error("expected hasAttachments") + } +} + +func TestM365SourceDeltaState(t *testing.T) { + graph := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"value":[{"id":"m1"}],"@odata.deltaLink":"` + deltaNext + `"}`)) + }) + c := newM365TestClient(t, graph) + deltaNext = c.base + "/v1.0/delta-next" + stateDir := filepath.Join(t.TempDir(), ".m365") + s := &m365Source{c: c, mailbox: "info@x.de", localpart: "info", stateDir: stateDir} + + if s.Folder() != "m365/info" { + t.Errorf("folder = %q", s.Folder()) + } + ids, _, err := s.ListIDs(context.Background(), 0, "") + if err != nil { + t.Fatalf("ListIDs: %v", err) + } + if len(ids) != 1 || ids[0] != "m1" { + t.Errorf("ids = %v", ids) + } + if err := s.Commit(); err != nil { + t.Fatalf("Commit: %v", err) + } + data, err := os.ReadFile(filepath.Join(stateDir, "info.deltalink")) + if err != nil { + t.Fatalf("read delta state: %v", err) + } + if string(data) != deltaNext { + t.Errorf("delta state = %q, want %q", string(data), deltaNext) + } +} + +func pathLast(p string) string { + for i := len(p) - 1; i >= 0; i-- { + if p[i] == '/' { + return p[i+1:] + } + } + return p +} + +// deltaNext/deltaFinal are set per-test from the fake graph server URL so +// deltaLink values always point back at the fake (never the real Graph). +var deltaNext, deltaFinal string diff --git a/bin/mail/sync/sync.go b/bin/mail/sync/sync.go index 2311278..13c11f8 100644 --- a/bin/mail/sync/sync.go +++ b/bin/mail/sync/sync.go @@ -106,6 +106,7 @@ func Retry(ctx context.Context, policy RetryPolicy, fn func() error) error { type SyncConfig struct { OO *OOConfig // OnlyOffice source (optional) Gmail *GmailCredentials // Gmail source (optional) + M365 *M365Credentials // Microsoft 365 Graph source (optional) Out string // var/mail root; default /var/mail Workers int // concurrency; default 4 Limit int // max messages per source (0 = all) @@ -133,6 +134,14 @@ type Source interface { Folder() string } +// Committer is an optional Source capability: Commit is called after all listed +// ids have been downloaded successfully. Sources that only advance durable state +// on success (e.g. a Graph delta link) implement this so a killed or failed run +// stays retryable without gaps. +type Committer interface { + Commit() error +} + type ooSource struct { c *OOClient page int @@ -222,8 +231,22 @@ func Run(ctx context.Context, cfg SyncConfig) (*SyncStats, error) { } sources = append(sources, &gmailSource{c: gm, query: cfg.Query}) } + if cfg.M365 != nil { + stateDir := filepath.Join(cfg.Out, ".m365") + for _, mb := range cfg.M365.Users { + if !strings.Contains(mb, "@") { + return nil, fmt.Errorf("m365 user %q is not an email address", mb) + } + c, err := NewM365Client(*cfg.M365) + if err != nil { + return nil, fmt.Errorf("m365 init for %s: %w", mb, err) + } + local := strings.SplitN(mb, "@", 2)[0] + sources = append(sources, &m365Source{c: c, mailbox: mb, localpart: strings.ToLower(local), stateDir: stateDir}) + } + } if len(sources) == 0 { - return nil, errors.New("sync: no source configured (need OO, Gmail, or both)") + return nil, errors.New("sync: no source configured (need OO, Gmail, M365, or a combination)") } stats := &SyncStats{} @@ -296,6 +319,25 @@ func Run(ctx context.Context, cfg SyncConfig) (*SyncStats, error) { close(jobsCh) wg.Wait() + // Only advance durable source state (e.g. delta links) when everything + // downloaded. A killed or failed run must be retryable without gaps. + if len(failures) == 0 { + seen := map[Source]bool{} + for _, j := range jobs { + if seen[j.src] { + continue + } + seen[j.src] = true + if c, ok := j.src.(Committer); ok { + if err := c.Commit(); err != nil { + mu.Lock() + failures = append(failures, j.src.Folder()+"/commit: "+err.Error()) + mu.Unlock() + } + } + } + } + if len(failures) > 0 { fmt.Fprintf(os.Stderr, "sync: %d failures:\n %s\n", len(failures), strings.Join(failures, "\n ")) } diff --git a/compose.yaml b/compose.yaml index 19475cc..434376f 100644 --- a/compose.yaml +++ b/compose.yaml @@ -92,6 +92,33 @@ services: tmpfs: - /tmp + # mail-sync: pull M365 (and OnlyOffice/Gmail) mail every 10 s into the shared + # var volume, then rebuild the brain index. Runs on the index image so it can + # convert + index in-process. Needs ~/.config/brain/m365.env for the M365 + # source; override MAIL_SYNC_SRC / MAIL_SYNC_ENV for other providers. + # docker compose up -d mail-sync + mail-sync: + image: ghcr.io/eslider/2dph:index + build: + context: . + dockerfile: Dockerfile + target: index + environment: + HF_HOME: /data/hf + KB_PY: python3 + MAIL_SYNC_SRC: m365 + MAIL_SYNC_ENV: /secret/m365.env + volumes: + - kb-model:/data/hf + - kb-var:/app/var + - ~/.config/brain:/secret:ro + command: ["mail-sync", "10"] + read_only: true + tmpfs: + - /tmp + restart: unless-stopped + stop_grace_period: 20s + # Optional local SearXNG (D3). Skip if BRAIN_SEARCH_URL already points at a # live instance — do not run a second copy on that host. # SEARXNG_SECRET=$(openssl rand -hex 32) docker compose --profile searxng up -d