feat(PROJ-21): Phase 2+3+5+8 Multi-Tenancy + PROJ-2 EML/MBOX Upload
Phase 2a: userstore domain_admin/superadmin Rollen, User.TenantID,
ListByTenant, UpsertLDAPUser mit tenantID
Phase 2b: storage.Save() mit tenantID *int64, email_refs Tabelle,
GetTenantForMail, GetAllIDsByTenant, StatsByTenant
Phase 2c: JWT-Claims tenant_id/tenant_slug, Session.TenantID,
Login Domain-Erkennung via E-Mail-Domain
Phase 3: tenantMiddleware, Handler-Filterung (Users, Mail, Stats)
Phase 5: SMTP Domain-Routing via DomainToTenantFunc Callback,
config smtp.tenant_routing + default_tenant_id
Phase 8: archivmail migrate-tenants Subkommando
PROJ-2: Upload-Seite /admin/upload mit DropZone + Progress-Polling
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
sysops
+160
-36
@@ -193,15 +193,33 @@ func (s *Store) initSchema(ctx context.Context) error {
|
||||
CREATE INDEX IF NOT EXISTS idx_emails_mail_from ON emails (mail_from);
|
||||
CREATE INDEX IF NOT EXISTS idx_emails_subject ON emails USING gin (to_tsvector('simple', subject));
|
||||
`)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Phase 2b migrations: tenant isolation
|
||||
_, err = s.db.Exec(ctx, `
|
||||
ALTER TABLE emails ADD COLUMN IF NOT EXISTS tenant_id BIGINT;
|
||||
CREATE INDEX IF NOT EXISTS idx_emails_tenant ON emails (tenant_id);
|
||||
CREATE TABLE IF NOT EXISTS email_refs (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
email_id TEXT NOT NULL REFERENCES emails(id),
|
||||
tenant_id BIGINT NOT NULL,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
UNIQUE(email_id, tenant_id)
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_email_refs_tenant ON email_refs (tenant_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_email_refs_email ON email_refs (email_id);
|
||||
`)
|
||||
return err
|
||||
}
|
||||
|
||||
// ── Core operations ───────────────────────────────────────────────────────
|
||||
|
||||
// Save writes raw email bytes to storage. The ID is the hex-encoded SHA256 of
|
||||
// the plaintext content. If the file already exists, Save is a no-op (dedup).
|
||||
// It also inserts metadata into the emails table if a DB is configured.
|
||||
func (s *Store) Save(raw []byte, _ time.Time) (string, error) {
|
||||
// the plaintext content. If the file already exists, Save ensures an email_ref
|
||||
// exists for the tenant (cross-tenant dedup: one file, many refs).
|
||||
// tenantID may be nil for system-level ingestion without tenant assignment.
|
||||
func (s *Store) Save(ctx context.Context, raw []byte, _ time.Time, tenantID *int64) (string, error) {
|
||||
// Hash plaintext for dedup (always before encryption)
|
||||
sum := sha256.Sum256(raw)
|
||||
id := fmt.Sprintf("%x", sum[:]) // 64 hex chars
|
||||
@@ -211,36 +229,46 @@ func (s *Store) Save(raw []byte, _ time.Time) (string, error) {
|
||||
return "", fmt.Errorf("storage: mkdir shard: %w", err)
|
||||
}
|
||||
|
||||
// Dedup: if file already exists, return same id
|
||||
fileExists := false
|
||||
if _, err := os.Stat(path); err == nil {
|
||||
return id, nil
|
||||
fileExists = true
|
||||
}
|
||||
|
||||
// Determine what to write: encrypted or plaintext
|
||||
var toWrite []byte
|
||||
if s.key != nil {
|
||||
encrypted, err := s.encrypt(raw)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
toWrite = encrypted
|
||||
} else {
|
||||
toWrite = raw
|
||||
}
|
||||
|
||||
if err := os.WriteFile(path, toWrite, 0o644); err != nil {
|
||||
return "", fmt.Errorf("storage: write: %w", err)
|
||||
}
|
||||
|
||||
// Insert metadata into DB (best-effort parse)
|
||||
if s.db != nil {
|
||||
pm, parseErr := mailparser.Parse(raw)
|
||||
if parseErr == nil {
|
||||
s.insertMeta(context.Background(), id, pm, len(raw))
|
||||
if !fileExists {
|
||||
// Determine what to write: encrypted or plaintext
|
||||
var toWrite []byte
|
||||
if s.key != nil {
|
||||
encrypted, err := s.encrypt(raw)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
toWrite = encrypted
|
||||
} else {
|
||||
// Insert minimal metadata even if parse fails
|
||||
s.insertMetaMinimal(context.Background(), id, len(raw))
|
||||
toWrite = raw
|
||||
}
|
||||
|
||||
if err := os.WriteFile(path, toWrite, 0o644); err != nil {
|
||||
return "", fmt.Errorf("storage: write: %w", err)
|
||||
}
|
||||
|
||||
// Insert metadata into DB (best-effort parse)
|
||||
if s.db != nil {
|
||||
pm, parseErr := mailparser.Parse(raw)
|
||||
if parseErr == nil {
|
||||
s.insertMeta(ctx, id, pm, len(raw), tenantID)
|
||||
} else {
|
||||
s.insertMetaMinimal(ctx, id, len(raw), tenantID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Ensure email_ref entry for this tenant (even if file already existed)
|
||||
if s.db != nil && tenantID != nil {
|
||||
_, _ = s.db.Exec(ctx, `
|
||||
INSERT INTO email_refs (email_id, tenant_id)
|
||||
VALUES ($1, $2)
|
||||
ON CONFLICT (email_id, tenant_id) DO NOTHING
|
||||
`, id, *tenantID)
|
||||
}
|
||||
|
||||
return id, nil
|
||||
@@ -402,24 +430,24 @@ func (s *Store) firstAndLastFromFS() (first, last *MailRef, err error) {
|
||||
// ── Metadata helpers ──────────────────────────────────────────────────────
|
||||
|
||||
// insertMeta inserts parsed email metadata into the emails table.
|
||||
func (s *Store) insertMeta(ctx context.Context, id string, pm *mailparser.ParsedMail, size int) {
|
||||
func (s *Store) insertMeta(ctx context.Context, id string, pm *mailparser.ParsedMail, size int, tenantID *int64) {
|
||||
mailTo := strings.Join(pm.To, ", ")
|
||||
hasAttach := len(pm.Attachments) > 0
|
||||
|
||||
_, _ = s.db.Exec(ctx, `
|
||||
INSERT INTO emails (id, received_at, mail_from, mail_to, subject, size_bytes, has_attach)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7)
|
||||
INSERT INTO emails (id, received_at, mail_from, mail_to, subject, size_bytes, has_attach, tenant_id)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
|
||||
ON CONFLICT (id) DO NOTHING
|
||||
`, id, pm.Date, pm.From, mailTo, pm.Subject, int64(size), hasAttach)
|
||||
`, id, pm.Date, pm.From, mailTo, pm.Subject, int64(size), hasAttach, tenantID)
|
||||
}
|
||||
|
||||
// insertMetaMinimal inserts minimal metadata when parsing fails.
|
||||
func (s *Store) insertMetaMinimal(ctx context.Context, id string, size int) {
|
||||
func (s *Store) insertMetaMinimal(ctx context.Context, id string, size int, tenantID *int64) {
|
||||
_, _ = s.db.Exec(ctx, `
|
||||
INSERT INTO emails (id, received_at, size_bytes)
|
||||
VALUES ($1, NOW(), $2)
|
||||
INSERT INTO emails (id, received_at, size_bytes, tenant_id)
|
||||
VALUES ($1, NOW(), $2, $3)
|
||||
ON CONFLICT (id) DO NOTHING
|
||||
`, id, int64(size))
|
||||
`, id, int64(size), tenantID)
|
||||
}
|
||||
|
||||
// SaveMeta upserts metadata for a given email ID. Used by the backfill process.
|
||||
@@ -602,6 +630,102 @@ func (s *Store) VerifyIntegrity(ctx context.Context, id string) (bool, error) {
|
||||
return ok, nil
|
||||
}
|
||||
|
||||
// GetTenantForMail returns the tenant_id stored directly on the email record.
|
||||
// Returns nil if no tenant is assigned or the mail does not exist.
|
||||
func (s *Store) GetTenantForMail(ctx context.Context, id string) (*int64, error) {
|
||||
if s.db == nil {
|
||||
return nil, nil
|
||||
}
|
||||
var tenantID *int64
|
||||
err := s.db.QueryRow(ctx, `SELECT tenant_id FROM emails WHERE id = $1`, id).Scan(&tenantID)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("storage: get tenant for mail: %w", err)
|
||||
}
|
||||
return tenantID, nil
|
||||
}
|
||||
|
||||
// GetAllIDsByTenant returns all email IDs visible to a tenant.
|
||||
// If tenantID is nil, all IDs are returned (superadmin / no-tenant context).
|
||||
func (s *Store) GetAllIDsByTenant(ctx context.Context, tenantID *int64) ([]string, error) {
|
||||
if s.db != nil {
|
||||
var (
|
||||
rows pgx.Rows
|
||||
err error
|
||||
)
|
||||
if tenantID == nil {
|
||||
rows, err = s.db.Query(ctx, `SELECT id FROM emails ORDER BY received_at`)
|
||||
} else {
|
||||
rows, err = s.db.Query(ctx,
|
||||
`SELECT email_id FROM email_refs WHERE tenant_id = $1`, *tenantID)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("storage: get ids by tenant: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var ids []string
|
||||
for rows.Next() {
|
||||
var id string
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
continue
|
||||
}
|
||||
ids = append(ids, id)
|
||||
}
|
||||
return ids, rows.Err()
|
||||
}
|
||||
// fallback: walk store (no tenant filtering possible without DB)
|
||||
var ids []string
|
||||
err := s.WalkStore(ctx, func(id string) error {
|
||||
ids = append(ids, id)
|
||||
return nil
|
||||
})
|
||||
return ids, err
|
||||
}
|
||||
|
||||
// StatsByTenant returns mail count and total size filtered by tenant.
|
||||
// If tenantID is nil, aggregate over all emails.
|
||||
func (s *Store) StatsByTenant(ctx context.Context, tenantID *int64) (map[string]interface{}, error) {
|
||||
if s.db == nil {
|
||||
st, err := s.statsFromFS()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]interface{}{
|
||||
"count": st.TotalMails,
|
||||
"total_size": st.TotalBytes,
|
||||
}, nil
|
||||
}
|
||||
|
||||
var count int64
|
||||
var totalSize int64
|
||||
|
||||
if tenantID == nil {
|
||||
err := s.db.QueryRow(ctx,
|
||||
`SELECT COALESCE(COUNT(*),0), COALESCE(SUM(size_bytes),0) FROM emails`,
|
||||
).Scan(&count, &totalSize)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("storage: stats by tenant: %w", err)
|
||||
}
|
||||
} else {
|
||||
err := s.db.QueryRow(ctx, `
|
||||
SELECT COALESCE(COUNT(e.id),0), COALESCE(SUM(e.size_bytes),0)
|
||||
FROM email_refs r
|
||||
JOIN emails e ON e.id = r.email_id
|
||||
WHERE r.tenant_id = $1
|
||||
`, *tenantID).Scan(&count, &totalSize)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("storage: stats by tenant: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
return map[string]interface{}{
|
||||
"count": count,
|
||||
"total_size": totalSize,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// GetAllIDs returns all email IDs from the DB, or walks the store if no DB.
|
||||
func (s *Store) GetAllIDs(ctx context.Context) ([]string, error) {
|
||||
if s.db != nil {
|
||||
|
||||
Reference in New Issue
Block a user