Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/missionmatch / internal/modules/recruiting/mailsync.go |
| Date | Not specified in source |
| Repository | Janady13/missionmatch |
| Revision | 720d47fae476b0d38477ae4401697d372b1ea2bc |
| Branch | feature/opportunity-to-outcome |
| Path or archive member | internal/modules/recruiting/mailsync.go |
| Original SHA-256 | d5c66989eec78fd7923da34a0d26204ca3e1b3af068b14b2782dd3f8b5c38b10 |
| Git blob SHA | 71a09abf8ddfaf73f15653b7609dc27cbfc76507 |
| Reading-copy SHA-256 | 113b514cdc07ed3ce8e3890108feee290b34697ed6d418f5233a0b456192de6d |
Recovered source transcript
Download complete source text ↓480 lines · 17.6 KB
Credential-bearing and unrelated personal details were removed from this reading copy. The source identity and original checksum remain recorded.
package recruiting
// MAIL SYNC. A recruiter connects their mailbox and their correspondence logs itself.
//
// recruiter_activities has had the right vocabulary since migration 0049 and every row in it had to be TYPED
// BY A HUMAN, which is the most reliable way to end up with an activity log nobody trusts: the people best at
// recruiting are the least likely to stop and write down that they sent an email.
//
// HEADERS ONLY, NEVER BODIES. The log needs who, when and the subject. Pulling bodies would drag a
// recruiter's entire mailbox, including correspondence with their doctor and their lawyer, into a platform
// holding controlled unclassified information where it would sit under the same retention rule as an
// employment decision record. The IMAP client cannot fetch a body: there is no parameter through which one
// could be requested.
//
// A MESSAGE THAT MATCHES NOBODY IS NOT STORED AT ALL. Every message is compared against this tenant's
// candidates and contacts by address and dropped if it matches none. A recruiter connecting their mailbox is
// not handing over their mailbox.
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/missionworkforce/missionmatch/internal/platform/audit"
"github.com/missionworkforce/missionmatch/internal/platform/imapclient"
"github.com/missionworkforce/missionmatch/internal/platform/orgsettings"
"github.com/missionworkforce/missionmatch/internal/platform/tenant"
)
var (
ErrMailboxNotConfigured = errors.New("recruiting: no mailbox is connected for this user")
ErrMailboxUnverified = errors.New("recruiting: this mailbox has not been verified, test the connection first")
// ErrMailInvalid is a CONFIGURATION mistake by the caller, distinct from a transport failure. Without a
// sentinel the handler could only pattern match on the message prefix, and every validation error shared
// that prefix with every connection error, so a missing host reported itself as a bad gateway.
ErrMailInvalid = errors.New("recruiting: mailbox configuration is invalid")
)
// MailAccount is one recruiter's mailbox connection. It never carries the password.
type MailAccount struct {
Host string `json:"imap_host"`
Port int `json:"imap_port"`
Username string `json:"imap_username"`
Mailbox string `json:"mailbox"`
SyncSent bool `json:"sync_sent"`
Enabled bool `json:"enabled"`
PasswordSet bool `json:"password_set"`
VerifiedAt *time.Time `json:"verified_at,omitempty"`
LastSyncAt *time.Time `json:"last_sync_at,omitempty"`
LastError string `json:"last_error,omitempty"`
LookbackDays int `json:"lookback_days"`
}
// SyncResult is what one run did. It reports SKIPPED and UNMATCHED separately, because "logged 4 of 300" is
// only alarming until you know that 296 were correspondence with people who are not in the system, which is
// the normal and desired case.
type SyncResult struct {
Examined int `json:"examined"`
Matched int `json:"matched"`
Logged int `json:"logged"`
Duplicate int `json:"already_known"`
Unmatched int `json:"unmatched_and_discarded"`
Note string `json:"note,omitempty"`
}
// GetMailAccount reads the caller's own mailbox configuration.
func (s *Service) GetMailAccount(ctx context.Context) (MailAccount, bool, error) {
var a MailAccount
if s.db == nil {
return a, false, ErrNoDatabase
}
actor := actorPtr(ctx)
if actor == nil {
return a, false, tenant.ErrNoTenant
}
found := false
err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
e := tx.QueryRow(ctx, `
SELECT imap_host, imap_port, imap_username, mailbox, sync_sent, enabled,
verified_at, last_sync_at, last_error, lookback_days
FROM mail_accounts WHERE user_id = $1`, *actor).
Scan(&a.Host, &a.Port, &a.Username, &a.Mailbox, &a.SyncSent, &a.Enabled,
&a.VerifiedAt, &a.LastSyncAt, &a.LastError, &a.LookbackDays)
if errors.Is(e, pgx.ErrNoRows) {
return nil
}
if e != nil {
return e
}
found = true
if s.settings != nil {
var n int
if e := tx.QueryRow(ctx, `SELECT count(*)::int FROM org_secrets WHERE name = $1`,
orgsettings.IMAPSecretName(*actor)).Scan(&n); e != nil {
return e
}
a.PasswordSet = n > 0
}
return nil
})
return a, found, err
}
// MailAccountInput configures the caller's OWN mailbox. There is deliberately no user parameter: a mailbox
// belongs to a person, and an endpoint that let one person configure another's mailbox would be a way to
// point somebody else's credentials at a server you control.
type MailAccountInput struct {
Host string `json:"imap_host"`
Port int `json:"imap_port"`
Username string `json:"imap_username"`
Password string `json:"password"`
Mailbox string `json:"mailbox"`
LookbackDays int `json:"lookback_days"`
Enabled *bool `json:"enabled"`
}
func (s *Service) SaveMailAccount(ctx context.Context, in MailAccountInput, ip, ua string) error {
if s.db == nil {
return ErrNoDatabase
}
if s.settings == nil {
return errors.New("recruiting: secret storage is not available on this node")
}
actor := actorPtr(ctx)
if actor == nil {
return tenant.ErrNoTenant
}
in.Host = strings.TrimSpace(in.Host)
in.Username = strings.TrimSpace(in.Username)
if in.Host == "" || len(in.Host) > 255 {
return fmt.Errorf("%w: a mail host is required", ErrMailInvalid)
}
if in.Port <= 0 {
in.Port = 993
}
if in.Port > 65535 {
return fmt.Errorf("%w: a port must be between 1 and 65535", ErrMailInvalid)
}
if in.Mailbox == "" {
in.Mailbox = "INBOX"
}
if len(in.Mailbox) > 200 {
return fmt.Errorf("%w: that mailbox name is too long", ErrMailInvalid)
}
if in.LookbackDays <= 0 {
in.LookbackDays = 30
}
if in.LookbackDays > 365 {
return fmt.Errorf("%w: a sync may look back at most a year", ErrMailInvalid)
}
return s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
// CHANGING THE CONNECTION CLEARS THE VERIFICATION and switches sync off, the same rule the mail
// sender follows. An edited host with a stale proof means the platform goes on claiming a mailbox
// works against a configuration nobody has tried.
if _, e := tx.Exec(ctx, `
INSERT INTO mail_accounts (organization_id, user_id, imap_host, imap_port, imap_username,
mailbox, lookback_days, enabled, verified_at)
VALUES (NULLIF(current_setting('app.current_org_id', true), '')::uuid, $1,$2,$3,$4,$5,$6,false,NULL)
ON CONFLICT (organization_id, user_id) DO UPDATE
SET imap_host = EXCLUDED.imap_host, imap_port = EXCLUDED.imap_port,
imap_username = EXCLUDED.imap_username, mailbox = EXCLUDED.mailbox,
lookback_days = EXCLUDED.lookback_days, enabled = false, verified_at = NULL,
last_error = '', updated_at = now()`,
*actor, in.Host, in.Port, in.Username, in.Mailbox, in.LookbackDays); e != nil {
return e
}
if strings.TrimSpace(in.Password) != "" {
if e := s.settings.PutSecret(ctx, tx, orgsettings.IMAPSecretName(*actor), in.Password, *actor); e != nil {
return e
}
}
if s.audit == nil {
return nil
}
// The host and the account name are recorded; the password is not, and never appears in this row.
body, _ := json.Marshal(map[string]any{"imap_host": in.Host, "mailbox": in.Mailbox})
return s.audit.OnTx(ctx, tx, audit.Event{
Action: "mail.account.configured", ActorUserID: actor,
EntityType: "mail_account", EntityID: actor.String(), NewValues: body,
IPAddress: ip, UserAgent: ua,
})
})
}
// DisconnectMailbox removes the connection, the stored credential, and the dedup ledger.
//
// THE LEDGER GOES TOO, deliberately. Leaving it would mean a recruiter who reconnects finds their history
// already marked as seen and gets nothing, with no way to clear it. The activities already logged are
// untouched, because those are a record of work that genuinely happened.
func (s *Service) DisconnectMailbox(ctx context.Context, ip, ua string) error {
if s.db == nil {
return ErrNoDatabase
}
actor := actorPtr(ctx)
if actor == nil {
return tenant.ErrNoTenant
}
return s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
if _, e := tx.Exec(ctx, `DELETE FROM synced_messages WHERE user_id = $1`, *actor); e != nil {
return e
}
if _, e := tx.Exec(ctx, `DELETE FROM mail_accounts WHERE user_id = $1`, *actor); e != nil {
return e
}
if s.settings != nil {
if e := s.settings.DeleteSecret(ctx, tx, orgsettings.IMAPSecretName(*actor)); e != nil {
return e
}
}
if s.audit == nil {
return nil
}
return s.audit.OnTx(ctx, tx, audit.Event{
Action: "mail.account.disconnected", ActorUserID: actor,
EntityType: "mail_account", EntityID: actor.String(), IPAddress: ip, UserAgent: ua,
})
})
}
// TestMailbox connects and selects the mailbox, proving the configuration before anything is enabled. It is
// the ONLY writer of verified_at, which is what makes that column mean what it says.
func (s *Service) TestMailbox(ctx context.Context) error {
cfg, mailbox, actor, err := s.mailConfig(ctx)
if err != nil {
return err
}
c, err := imapclient.Dial(cfg)
if err == nil {
err = c.Select(mailbox)
_ = c.Close()
}
// THE OUTCOME IS RECORDED FIRST AND THE ERROR IS RETURNED AFTER, outside the transaction.
//
// This was written the other way round and it silently discarded every failure: returning the error from
// inside RunInTenantTx rolls the transaction back, taking with it the row that recorded WHY the test
// failed. So a user saw a 502 and then found last_error empty, which is the least helpful possible
// combination. The auth module has the same note about an audit row for the same reason.
var recorded string
if err != nil {
recorded = scrub(err.Error(), cfg.Username, cfg.Password)
if len(recorded) > 400 {
recorded = recorded[:400]
}
}
if e := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
if err != nil {
_, ex := tx.Exec(ctx, `UPDATE mail_accounts SET verified_at = NULL, enabled = false,
last_error = $2, updated_at = now() WHERE user_id = $1`, actor, recorded)
return ex
}
_, ex := tx.Exec(ctx, `UPDATE mail_accounts SET verified_at = now(), last_error = '',
updated_at = now() WHERE user_id = $1`, actor)
return ex
}); e != nil {
return e
}
if err != nil {
return fmt.Errorf("recruiting: %s", recorded)
}
return nil
}
// SetMailboxEnabled switches sync on or off. On is refused unless the connection has been proven.
func (s *Service) SetMailboxEnabled(ctx context.Context, on bool) error {
if s.db == nil {
return ErrNoDatabase
}
actor := actorPtr(ctx)
if actor == nil {
return tenant.ErrNoTenant
}
return s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
tag, e := tx.Exec(ctx, `UPDATE mail_accounts SET enabled = $2, updated_at = now()
WHERE user_id = $1 AND ($2 = false OR verified_at IS NOT NULL)`, actor, on)
if e != nil {
return e
}
if tag.RowsAffected() == 0 {
return ErrMailboxUnverified
}
return nil
})
}
// SyncMailbox runs one pass.
func (s *Service) SyncMailbox(ctx context.Context) (SyncResult, error) {
var res SyncResult
cfg, mailbox, actor, err := s.mailConfig(ctx)
if err != nil {
return res, err
}
var lookback int
var enabled bool
if e := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
return tx.QueryRow(ctx, `SELECT lookback_days, enabled FROM mail_accounts WHERE user_id = $1`,
actor).Scan(&lookback, &enabled)
}); e != nil {
return res, e
}
if !enabled {
return res, ErrMailboxUnverified
}
c, err := imapclient.Dial(cfg)
if err != nil {
return res, fmt.Errorf("recruiting: %s", scrub(err.Error(), cfg.Username, cfg.Password))
}
defer c.Close()
if err := c.Select(mailbox); err != nil {
return res, fmt.Errorf("recruiting: %s", scrub(err.Error(), cfg.Username, cfg.Password))
}
uids, err := c.SearchSince(time.Now().AddDate(0, 0, -lookback))
if err != nil {
return res, fmt.Errorf("recruiting: %s", scrub(err.Error(), cfg.Username, cfg.Password))
}
headers, err := c.FetchHeaders(uids)
if err != nil && len(headers) == 0 {
return res, fmt.Errorf("recruiting: %s", scrub(err.Error(), cfg.Username, cfg.Password))
}
res.Examined = len(headers)
err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
mine := strings.ToLower(cfg.Username)
for _, h := range headers {
// Every address on the message except the recruiter's own. The recruiter is not a candidate.
addrs := make([]string, 0, 8)
for _, a := range append(append([]string{h.From}, h.To...), h.Cc...) {
if a != "" && a != mine {
addrs = append(addrs, a)
}
}
if len(addrs) == 0 {
res.Unmatched++
continue
}
kind, id, name, ok, e := matchParty(ctx, tx, addrs)
if e != nil {
return e
}
if !ok {
// NOT STORED AT ALL. This is the promise: correspondence with somebody who is not in the
// system leaves no trace here.
res.Unmatched++
continue
}
res.Matched++
direction := "inbound"
if strings.EqualFold(h.From, mine) {
direction = "outbound"
}
occurred := h.Date
if occurred.IsZero() {
occurred = time.Now().UTC()
}
var activityID int64
e = tx.QueryRow(ctx, `
INSERT INTO recruiter_activities
(organization_id, actor_user_id, kind, direction, outcome, candidate_id, contact_id,
body, auto_captured, occurred_at)
VALUES (NULLIF(current_setting('app.current_org_id', true), '')::uuid,
$1,'email',$2,'logged',$3,$4,$5,true,$6)
RETURNING id`,
actor, direction,
nullIf(kind == "candidate", id), nullIf(kind == "contact", id),
subjectLine(h.Subject, name), occurred).Scan(&activityID)
if e != nil {
return e
}
// The dedup ledger is written LAST and its conflict is what makes a re run idempotent. A message
// seen before conflicts here, the insert is skipped, and the activity written a moment ago is
// rolled back with it, because both are in one transaction.
tag, e := tx.Exec(ctx, `
INSERT INTO synced_messages (organization_id, message_id, user_id, direction,
matched_kind, matched_id, activity_id)
VALUES (NULLIF(current_setting('app.current_org_id', true), '')::uuid, $1,$2,$3,$4,$5,$6)
ON CONFLICT (organization_id, message_id) DO NOTHING`,
h.MessageID, actor, direction, kind, id, activityID)
if e != nil {
return e
}
if tag.RowsAffected() == 0 {
// Already known. Undo the activity row just written for it.
if _, e := tx.Exec(ctx, `DELETE FROM recruiter_activities WHERE id = $1`, activityID); e != nil {
return e
}
res.Duplicate++
res.Matched--
continue
}
res.Logged++
}
_, e := tx.Exec(ctx, `UPDATE mail_accounts SET last_sync_at = now(), last_error = ''
WHERE user_id = $1`, actor)
return e
})
if res.Unmatched > 0 {
res.Note = "messages matching nobody in this organization were discarded and not stored"
}
return res, err
}
// matchParty finds the first candidate or contact holding any of these addresses. Candidates are checked
// first because they are the larger and more frequently correspondent population.
func matchParty(ctx context.Context, tx pgx.Tx, addrs []string) (kind string, id uuid.UUID, name string, ok bool, err error) {
err = tx.QueryRow(ctx, `SELECT id, full_name FROM candidates
WHERE lower(email) = ANY($1) AND deleted_at IS NULL LIMIT 1`, addrs).Scan(&id, &name)
if err == nil {
return "candidate", id, name, true, nil
}
if !errors.Is(err, pgx.ErrNoRows) {
return "", uuid.Nil, "", false, err
}
err = tx.QueryRow(ctx, `SELECT id, coalesce(full_name,'') FROM client_contacts
WHERE lower(email) = ANY($1) LIMIT 1`, addrs).Scan(&id, &name)
if err == nil {
return "contact", id, name, true, nil
}
if !errors.Is(err, pgx.ErrNoRows) {
return "", uuid.Nil, "", false, err
}
return "", uuid.Nil, "", false, nil
}
// subjectLine is what a colleague reads in the activity stream. The subject alone is often meaningless out of
// context, so the matched person's name is carried with it.
func subjectLine(subject, who string) string {
subject = strings.TrimSpace(subject)
if subject == "" {
subject = "(no subject)"
}
if who != "" {
return subject + "\n\nSynced from mail, correspondence with " + who + "."
}
return subject + "\n\nSynced from mail."
}
// mailConfig assembles the connection for the CALLING user, opening the sealed password. It is the only
// place that password is in memory, and it is never returned to a caller.
func (s *Service) mailConfig(ctx context.Context) (imapclient.Config, string, uuid.UUID, error) {
var cfg imapclient.Config
var mailbox string
if s.db == nil {
return cfg, "", uuid.Nil, ErrNoDatabase
}
if s.settings == nil {
return cfg, "", uuid.Nil, errors.New("recruiting: secret storage is not available on this node")
}
actor := actorPtr(ctx)
if actor == nil {
return cfg, "", uuid.Nil, tenant.ErrNoTenant
}
err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
e := tx.QueryRow(ctx, `SELECT imap_host, imap_port, imap_username, mailbox
FROM mail_accounts WHERE user_id = $1`, *actor).
Scan(&cfg.Host, &cfg.Port, &cfg.Username, &mailbox)
if errors.Is(e, pgx.ErrNoRows) {
return ErrMailboxNotConfigured
}
if e != nil {
return e
}
pw, e := s.settings.OpenSecret(ctx, tx, orgsettings.IMAPSecretName(*actor))
if e != nil {
if errors.Is(e, orgsettings.ErrNoSecret) {
return errors.New("recruiting: no mailbox password is [credential removed]")
}
return e
}
cfg.Password = pw
return nil
})
return cfg, mailbox, *actor, err
}