archivmail/cmd/archivmail/cmd_index_pending.go

package main

import (
	"context"
	"flag"
	"log/slog"
	"os"
	"strings"
	"time"

	"archivmail/config"
	"archivmail/internal/index"
	"archivmail/internal/storage"
	"archivmail/pkg/mailparser"
)

// runIndexPending indexes all mails that have not yet been indexed
// (indexed_at IS NULL). It loads matching IDs from the DB, builds a
// MailDocument per mail, queues them on a TenantIndexWorker, waits for the
// worker to drain, then exits. Designed to be driven by cron when
// index.batch_mode is enabled (PROJ-58).
//
// Usage:
//
//	archivmail index-pending --config /etc/archivmail/config.yml
//	archivmail index-pending --limit 500
func runIndexPending(args []string) {
	fs := flag.NewFlagSet("index-pending", flag.ExitOnError)
	configPath := fs.String("config", "/etc/archivmail/config.yml", "path to config file")
	limitFlag := fs.Int("limit", 0, "max number of mails to index (0 = no limit)")
	fs.Parse(args)

	logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))

	cfg, err := config.Load(*configPath)
	if err != nil {
		logger.Error("failed to load config", "err", err)
		os.Exit(1)
	}

	storeCfg := storage.Config{
		Dir:             cfg.Storage.StorePath,
		Keyfile:         cfg.Storage.Keyfile,
		DSN:             cfg.Database.DSN(),
		CompressEnabled: cfg.Storage.Compress,
	}
	mailStore, err := storage.New(storeCfg)
	if err != nil {
		logger.Error("storage init failed", "err", err)
		os.Exit(1)
	}
	defer mailStore.Close()

	indexBackend := cfg.Index.Backend
	if indexBackend == "" {
		indexBackend = "manticore"
	}

	batchSize := cfg.Index.BatchSize
	if batchSize <= 0 {
		batchSize = 100
	}

	var idxMgr index.TenantIndexer
	if indexBackend == "manticore" {
		dsn := cfg.Index.ManticoreDSN
		if dsn == "" {
			dsn = "manticore@tcp(127.0.0.1:9306)/?charset=utf8mb4"
		}
		m, err := index.NewManticoreTenantManager(dsn)
		if err != nil {
			logger.Error("manticore init failed", "err", err)
			os.Exit(1)
		}
		idxMgr = m
	} else {
		m, err := index.NewTenantIndexManager(cfg.Index.Path, batchSize, indexBackend)
		if err != nil {
			logger.Error("index manager init failed", "err", err)
			os.Exit(1)
		}
		idxMgr = m
	}
	defer idxMgr.Close()

	ctx := context.Background()

	mails, err := mailStore.GetUnindexedMails(ctx, *limitFlag)
	if err != nil {
		logger.Error("failed to list unindexed mails", "err", err)
		os.Exit(1)
	}

	logger.Info("index-pending: starting", "count", len(mails), "limit", *limitFlag)
	if len(mails) == 0 {
		return
	}

	// Queue size needs to fit the entire batch so Submit never drops.
	qSize := len(mails) + 16
	worker := index.NewTenantWorker(idxMgr, qSize, logger)
	worker.Start()

	// Periodic progress while waiting for the queue to drain.
	tick := time.NewTicker(5 * time.Second)
	defer tick.Stop()
	go func() {
		for range tick.C {
			logger.Info("index-pending: progress", "queue_remaining", worker.QueueLen())
		}
	}()

	submitted := 0
	for _, m := range mails {
		raw, err := mailStore.Load(m.ID)
		if err != nil {
			logger.Warn("index-pending: load failed", "id", m.ID, "err", err)
			continue
		}
		pm, err := mailparser.Parse(raw)
		if err != nil {
			logger.Warn("index-pending: parse failed, skipping", "id", m.ID, "err", err)
			continue
		}

		var attachNames []string
		for _, a := range pm.Attachments {
			if a.Filename != "" {
				attachNames = append(attachNames, a.Filename)
			}
		}

		doc := index.MailDocument{
			ID:            m.ID,
			From:          pm.From,
			To:            strings.Join(pm.To, ", "),
			CC:            strings.Join(pm.CC, ", "),
			Subject:       pm.Subject,
			Body:          pm.TextBody,
			AttachNames:   strings.Join(attachNames, " "),
			HasAttachment: len(pm.Attachments) > 0,
			Date:          pm.Date,
			Size:          int64(len(raw)),
			TenantID:      m.TenantID,
		}
		worker.Submit(doc)

		// Mark as indexed in DB so subsequent runs skip it.
		if err := mailStore.SetIndexedAt(ctx, m.ID); err != nil {
			logger.Warn("index-pending: set indexed_at failed", "id", m.ID, "err", err)
		}
		submitted++
	}

	worker.Stop() // waits for the queue to drain
	logger.Info("index-pending: complete", "submitted", submitted)
}