feat: HA scheduler leadership + templated email log

HA scheduler: the job scheduler and alert evaluator hold session-scoped
PostgreSQL advisory locks on pinned pool connections — replicas scan for
due work exactly once, failover verified live with two replicas. Manual
and hook-triggered runs stay available on any replica.

Email surface: mailer renders registered templates (invitation); every
send is logged to core.email_log with status and error text. New
GET /projects/:id/emails endpoint; Members page shows an Email log card.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

apps/backend/internal/services/alerting.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
 
 	db "github.com/tdvorak/primora/apps/backend/internal/database/db"
 	"github.com/tdvorak/primora/apps/backend/internal/models"
@@ -239,10 +240,11 @@ type alertEvaluator struct {
 	quit    chan struct{}
 	wg      sync.WaitGroup
 	started sync.Once
+	leader  *leaderState
 }
 
-func newAlertEvaluator(q alertQueries, publish func(ctx context.Context, projectID uuid.UUID, eventType string, data map[string]any), logger *slog.Logger) *alertEvaluator {
-	return &alertEvaluator{q: q, publish: publish, logger: logger, tick: time.Minute, quit: make(chan struct{})}
+func newAlertEvaluator(q alertQueries, publish func(ctx context.Context, projectID uuid.UUID, eventType string, data map[string]any), logger *slog.Logger, pool *pgxpool.Pool) *alertEvaluator {
+	return &alertEvaluator{q: q, publish: publish, logger: logger, tick: time.Minute, quit: make(chan struct{}), leader: newLeaderState(pool, leaderKeyAlerts)}
 }
 
 func (e *alertEvaluator) Start(ctx context.Context) {
@@ -255,8 +257,11 @@ func (e *alertEvaluator) Start(ctx context.Context) {
 			for {
 				select {
 				case <-t.C:
-					e.scan(ctx)
+					if e.leader.hold(ctx) {
+						e.scan(ctx)
+					}
 				case <-e.quit:
+					e.leader.stop(context.Background())
 					return
 				case <-ctx.Done():
 					return

apps/backend/internal/services/jobs.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/robfig/cron/v3"
 
 	db "github.com/tdvorak/primora/apps/backend/internal/database/db"
@@ -406,6 +407,7 @@ type JobScheduler struct {
 	quit     chan struct{}
 	wg       sync.WaitGroup
 	started  sync.Once
+	leader   *leaderState
 	// onEvent fans a finished run out to webhooks and the realtime stream.
 	onEvent func(ctx context.Context, projectID uuid.UUID, eventType string, data map[string]any)
 }
@@ -415,7 +417,7 @@ type jobExec struct {
 	run db.CoreScheduledJobRun
 }
 
-func NewJobScheduler(q jobQueries, enc *secrets.Encryptor, logger *slog.Logger, onEvent func(ctx context.Context, projectID uuid.UUID, eventType string, data map[string]any)) *JobScheduler {
+func NewJobScheduler(q jobQueries, enc *secrets.Encryptor, logger *slog.Logger, onEvent func(ctx context.Context, projectID uuid.UUID, eventType string, data map[string]any), pool *pgxpool.Pool) *JobScheduler {
 	if logger == nil {
 		logger = slog.Default()
 	}
@@ -428,6 +430,7 @@ func NewJobScheduler(q jobQueries, enc *secrets.Encryptor, logger *slog.Logger,
 		queue:    make(chan jobExec, 256),
 		quit:     make(chan struct{}),
 		onEvent:  onEvent,
+		leader:   newLeaderState(pool, leaderKeyScheduler),
 	}
 }
 
@@ -452,8 +455,13 @@ func (j *JobScheduler) run(ctx context.Context) {
 		case exec := <-j.queue:
 			j.execute(exec.job, exec.run)
 		case <-ticker.C:
-			j.scan()
+			// Only the leader fires scheduled runs — manual/hook enqueues are
+			// already scoped to whichever replica received the request.
+			if j.leader.hold(ctx) {
+				j.scan()
+			}
 		case <-j.quit:
+			j.leader.stop(context.Background())
 			return
 		case <-ctx.Done():
 			return

apps/backend/internal/services/leader.go

@@ -0,0 +1,98 @@
+package services
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// Leader election via Postgres advisory locks. The lock is session-scoped, so
+// the lease pins its pool connection for its entire lifetime — a handoff back
+// to the pool would silently drop it. A single-replica deployment simply
+// always holds the lock; there is no feature flag.
+//
+// Distinct keys per component so the job scheduler and the alert evaluator
+// elect independently.
+const (
+	leaderKeyScheduler int64 = 728201
+	leaderKeyAlerts    int64 = 728202
+)
+
+type leaderLease struct {
+	conn *pgxpool.Conn
+	key  int64
+}
+
+// acquireLeadership returns a lease when this process wins the lock, nil when
+// another replica holds it, or an error when the check itself failed.
+func acquireLeadership(ctx context.Context, pool *pgxpool.Pool, key int64) (*leaderLease, error) {
+	conn, err := pool.Acquire(ctx)
+	if err != nil {
+		return nil, err
+	}
+	var got bool
+	if err := conn.QueryRow(ctx, "SELECT pg_try_advisory_lock($1)", key).Scan(&got); err != nil {
+		conn.Release()
+		return nil, err
+	}
+	if !got {
+		conn.Release()
+		return nil, nil
+	}
+	return &leaderLease{conn: conn, key: key}, nil
+}
+
+// alive pings the pinned connection — a dead conn means Postgres dropped the
+// session and released the lock, so leadership is lost even though the lease
+// object still exists.
+func (l *leaderLease) alive(ctx context.Context) bool {
+	var one int
+	return l.conn.QueryRow(ctx, "SELECT 1").Scan(&one) == nil
+}
+
+func (l *leaderLease) release(ctx context.Context) {
+	_, _ = l.conn.Exec(ctx, "SELECT pg_advisory_unlock($1)", l.key)
+	l.conn.Release()
+}
+
+// leaderState tracks a lease inside a ticker loop — acquire when absent,
+// verify when held, drop and re-elect when the connection dies.
+type leaderState struct {
+	pool  *pgxpool.Pool
+	key   int64
+	lease *leaderLease
+}
+
+func newLeaderState(pool *pgxpool.Pool, key int64) *leaderState {
+	return &leaderState{pool: pool, key: key}
+}
+
+// hold reports whether this process currently leads. A nil pool means no
+// election configured (tests, embedded use) — always lead.
+func (l *leaderState) hold(ctx context.Context) bool {
+	if l == nil || l.pool == nil {
+		return true
+	}
+	if l.lease != nil {
+		if l.lease.alive(ctx) {
+			return true
+		}
+		l.lease.release(ctx)
+		l.lease = nil
+	}
+	lease, err := acquireLeadership(ctx, l.pool, l.key)
+	if err != nil {
+		return false
+	}
+	l.lease = lease
+	return lease != nil
+}
+
+func (l *leaderState) stop(ctx context.Context) {
+	if l != nil && l.lease != nil {
+		l.lease.release(ctx)
+		l.lease = nil
+	}
+}
+
+

apps/backend/internal/services/mailer.go

@@ -11,6 +11,9 @@ import (
 	"github.com/tdvorak/primora/apps/backend/internal/config"
 )
 
+// Mailer is transport-only: Resend when configured, plain SMTP otherwise.
+// Templates render (subject, body) from data; sends are logged to
+// core.email_log by the service layer so operators can see what went out.
 type Mailer struct {
 	cfg    config.Config
 	resend *resend.Client
@@ -24,9 +27,30 @@ func NewMailer(cfg config.Config) *Mailer {
 	return &Mailer{cfg: cfg, resend: resendClient}
 }
 
-func (m *Mailer) SendInvitation(ctx context.Context, toEmail, organizationName, inviteURL string) error {
-	subject := fmt.Sprintf("You were invited to %s on Primora", organizationName)
-	text := "You have been invited to Primora.\n\nOpen this link to accept the invitation:\n" + inviteURL + "\n"
+// mailTemplate renders a subject + plaintext body from template data.
+type mailTemplate func(data map[string]string) (subject string, body string)
+
+var mailTemplates = map[string]mailTemplate{
+	"invitation": func(d map[string]string) (string, string) {
+		subject := fmt.Sprintf("You were invited to %s on Primora", d["organization"])
+		body := fmt.Sprintf(
+			"You have been invited to Primora.\n\nOpen this link to accept the invitation:\n%s\n\nThis link expires in 72 hours.\n",
+			d["invite_url"],
+		)
+		return subject, body
+	},
+}
+
+func renderMailTemplate(template string, data map[string]string) (string, string, error) {
+	tpl, ok := mailTemplates[template]
+	if !ok {
+		return "", "", fmt.Errorf("unknown email template %q", template)
+	}
+	subject, body := tpl(data)
+	return subject, body, nil
+}
+
+func (m *Mailer) Send(ctx context.Context, toEmail, subject, text string) error {
 	if m.resend != nil {
 		_, err := m.resend.Emails.SendWithContext(ctx, &resend.SendEmailRequest{
 			From:    m.cfg.MailFrom,

apps/backend/internal/services/mailer_test.go

@@ -0,0 +1,37 @@
+package services
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestRenderMailTemplateInvitation(t *testing.T) {
+	subject, body, err := renderMailTemplate("invitation", map[string]string{
+		"organization": "Acme Corp",
+		"invite_url":   "https://app.primora.dev/accept?token=abc",
+	})
+	if err != nil {
+		t.Fatalf("render: %v", err)
+	}
+	if !strings.Contains(subject, "Acme Corp") {
+		t.Fatalf("subject missing organization: %q", subject)
+	}
+	if !strings.Contains(body, "https://app.primora.dev/accept?token=abc") {
+		t.Fatalf("body missing invite url: %q", body)
+	}
+}
+
+func TestRenderMailTemplateUnknown(t *testing.T) {
+	if _, _, err := renderMailTemplate("does-not-exist", nil); err == nil {
+		t.Fatal("expected error for unknown template")
+	}
+}
+
+func TestExtractEmail(t *testing.T) {
+	if got := extractEmail("Primora <[email protected]>"); got != "[email protected]" {
+		t.Fatalf("extractEmail = %q", got)
+	}
+	if got := extractEmail("[email protected]"); got != "[email protected]" {
+		t.Fatalf("extractEmail bare = %q", got)
+	}
+}

apps/backend/internal/services/platform_service.go

@@ -229,14 +229,78 @@ func NewPlatformService(repo *repositories.CoreRepository, store *storage.LocalS
 	if enc != nil {
 		s.dispatcher = NewWebhookDispatcher(repo.Queries(), enc, logger)
 		s.dispatcher.Start(context.Background())
-		s.scheduler = NewJobScheduler(repo.Queries(), enc, logger, s.publishEvent)
+		s.scheduler = NewJobScheduler(repo.Queries(), enc, logger, s.publishEvent, repo.Pool())
 		s.scheduler.Start(context.Background())
-		s.alerts = newAlertEvaluator(repo.Queries(), s.publishEvent, logger)
+		s.alerts = newAlertEvaluator(repo.Queries(), s.publishEvent, logger, repo.Pool())
 		s.alerts.Start(context.Background())
 	}
 	return s
 }
 
+// sendTemplatedEmail renders a registered template, sends it, and writes an
+// email_log row with the outcome — failed sends are still logged so the
+// dashboard shows them.
+func (s *PlatformService) sendTemplatedEmail(ctx context.Context, projectID *uuid.UUID, template, to string, data map[string]string) error {
+	subject, body, err := renderMailTemplate(template, data)
+	status, errText := "sent", ""
+	if err == nil {
+		if sendErr := s.mailer.Send(ctx, to, subject, body); sendErr != nil {
+			status, errText, err = "failed", sendErr.Error(), sendErr
+		}
+	} else {
+		status, errText = "failed", err.Error()
+	}
+	var pid pgtype.UUID
+	if projectID != nil {
+		pid = pgtype.UUID{Bytes: *projectID, Valid: true}
+	}
+	_, _ = s.repo.Queries().InsertEmailLog(ctx, db.InsertEmailLogParams{
+		ProjectID: pid,
+		Template:  template,
+		ToEmail:   to,
+		Subject:   subject,
+		Status:    status,
+		Error:     errText,
+	})
+	return err
+}
+
+type EmailLogSummary struct {
+	ID        string    `json:"id"`
+	Template  string    `json:"template"`
+	ToEmail   string    `json:"to_email"`
+	Subject   string    `json:"subject"`
+	Status    string    `json:"status"`
+	Error     string    `json:"error,omitempty"`
+	CreatedAt time.Time `json:"created_at"`
+}
+
+func (s *PlatformService) ListEmailLog(ctx context.Context, actor *models.Actor, projectID uuid.UUID, limit int64) ([]EmailLogSummary, error) {
+	if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer", "viewer"); err != nil {
+		return nil, err
+	}
+	if limit <= 0 || limit > 200 {
+		limit = 50
+	}
+	rows, err := s.repo.Queries().ListEmailLog(ctx, db.ListEmailLogParams{ProjectID: pgtype.UUID{Bytes: projectID, Valid: true}, Limit: int32(limit)})
+	if err != nil {
+		return nil, err
+	}
+	out := make([]EmailLogSummary, len(rows))
+	for i, r := range rows {
+		out[i] = EmailLogSummary{
+			ID:        r.ID.String(),
+			Template:  r.Template,
+			ToEmail:   r.ToEmail,
+			Subject:   r.Subject,
+			Status:    r.Status,
+			Error:     r.Error,
+			CreatedAt: r.CreatedAt.Time,
+		}
+	}
+	return out, nil
+}
+
 // Close stops the webhook dispatcher and job scheduler. Idempotent.
 func (s *PlatformService) Close() {
 	if s.dispatcher != nil {
@@ -913,7 +977,15 @@ func (s *PlatformService) CreateInvitation(ctx context.Context, actor *models.Ac
 	if err != nil {
 		inviteURL = strings.TrimRight(baseInviteURL, "/") + "/" + token
 	}
-	if err := s.mailer.SendInvitation(ctx, invitation.Email, "Primora", inviteURL); err != nil {
+	var inviteProject *uuid.UUID
+	if projectUUID.Valid {
+		id := uuid.UUID(projectUUID.Bytes)
+		inviteProject = &id
+	}
+	if err := s.sendTemplatedEmail(ctx, inviteProject, "invitation", invitation.Email, map[string]string{
+		"organization": "Primora",
+		"invite_url":   inviteURL,
+	}); err != nil {
 		return nil, err
 	}
 	return map[string]any{