website/server/src/model/engagement/engagementRecipients.db.js

const { query } = require('../../utils/db')

// A bound on every "resolve an audience" query. `authenticated` on a large
// deployment is the whole user table, and the engine turns each id into an
// outbox row - so the read that feeds it has to have a ceiling of its own. The
// per-rule hourly cap (§7.1 Q3) is the operator-facing limit; this is the one
// that keeps a single emit from loading a hundred thousand rows into memory.
const MAX_AUDIENCE = 5000

const ids = (rows) => rows.map((r) => Number(r.id)).filter((n) => Number.isInteger(n) && n > 0)

const marks = (list) => list.map(() => '?').join(', ')

/**
 * Every active user. The `authenticated` audience - and `everyone`, which has no
 * distinct meaning here: a signed-out visitor has no address, no device and no
 * inbox, so the widest set the engine can actually deliver to is this one. The
 * ceiling lattice still distinguishes them (a trigger ceilinged `everyone`
 * permits an `authenticated` rule and not the reverse); only the resolution
 * coincides.
 *
 * `status = 'active'` on every query in this file: a banned or disabled account
 * is refused at login, and mailing it engagement content would be the one
 * surface that did not get the message.
 */
const active = async (limit = MAX_AUDIENCE) =>
  ids(await query("SELECT id FROM users WHERE status = 'active' ORDER BY id LIMIT ?", [limit]))

/** The `staff` audience. Roles come from `ceilings.STAFF_CEILING_ROLES`. */
const staff = async (roles, limit = MAX_AUDIENCE) => {
  if (!roles.length) return []
  return ids(
    await query(
      `SELECT id FROM users WHERE status = 'active' AND role IN (${marks(roles)}) ORDER BY id LIMIT ?`,
      [...roles, limit],
    ),
  )
}

/**
 * The `subscribers` audience: active users who have opted into this id on at
 * least one channel.
 *
 * "Opted in" is the EFFECTIVE mode, not the stored one, and that is why this is
 * not simply `WHERE mode <> 'off'`. A row exists only where a user said
 * something; absence means the channel's `defaultMode` (§3.1). All three of
 * core's channels default 'off' today, so the second half of the WHERE matches
 * nobody - but writing it means the day a channel ships with a non-off default,
 * this audience is already right rather than silently excluding everyone who
 * never opened the preferences screen.
 *
 * `defaultOnChannels` is the caller's list of channels whose defaultMode is not
 * 'off'; it comes from the channel registry, so the default lives in exactly one
 * place here too.
 */
const subscribers = async (streamId, defaultOnChannels = [], limit = MAX_AUDIENCE) => {
  const optedIn = `EXISTS (
      SELECT 1 FROM notification_channel_prefs p
       WHERE p.user_id = u.id AND p.stream_id = ? AND p.mode <> 'off')`

  if (!defaultOnChannels.length) {
    return ids(
      await query(
        `SELECT u.id FROM users u WHERE u.status = 'active' AND ${optedIn} ORDER BY u.id LIMIT ?`,
        [streamId, limit],
      ),
    )
  }

  // "At least one default-on channel has no row for this user" - counted rather
  // than NOT EXISTS, because NOT EXISTS would mean "none of them has a row".
  const defaulted = `(
      SELECT COUNT(*) FROM notification_channel_prefs p2
       WHERE p2.user_id = u.id AND p2.stream_id = ? AND p2.channel IN (${marks(defaultOnChannels)})
    ) < ?`

  return ids(
    await query(
      `SELECT u.id FROM users u
        WHERE u.status = 'active' AND (${optedIn} OR ${defaulted})
        ORDER BY u.id LIMIT ?`,
      [streamId, streamId, ...defaultOnChannels, defaultOnChannels.length, limit],
    ),
  )
}

/**
 * Narrow a set of user ids to the active ones.
 *
 * Every audience that does NOT come from a query in this file goes through here:
 * `owner` is a single id off the event envelope, and a module-declared audience
 * (§5.1a) is a list of ids a module's own resolver produced. Neither has any
 * notion of account status, and a module must not be able to mail a banned
 * account by returning its id.
 */
const filterActive = async (userIds) => {
  const wanted = [...new Set(userIds.map(Number).filter((n) => Number.isInteger(n) && n > 0))]
  if (!wanted.length) return []
  const capped = wanted.slice(0, MAX_AUDIENCE)
  return ids(
    await query(
      `SELECT id FROM users WHERE status = 'active' AND id IN (${marks(capped)}) ORDER BY id`,
      capped,
    ),
  )
}

/**
 * The stored mode for one (id, channel) across a set of users, as a Map.
 *
 * The caller applies the channel's `defaultMode` to anyone missing from the map,
 * which keeps the defaulting in the one place §3.1 put it. Returning stored rows
 * rather than a decision is what makes that possible.
 */
const storedModes = async (userIds, streamId, channel) => {
  if (!userIds.length) return new Map()
  const rows = await query(
    `SELECT user_id, mode FROM notification_channel_prefs
      WHERE stream_id = ? AND channel = ? AND user_id IN (${marks(userIds)})`,
    [streamId, channel, ...userIds],
  )
  return new Map(rows.map((r) => [Number(r.user_id), r.mode]))
}

/**
 * One user's mailable address, or null — the email channel's `addressFor`
 * (Phase 6).
 *
 * `status = 'active'` is re-checked here even though every audience query already
 * filtered on it, and the gap it closes is real rather than theoretical: an
 * outbox row can sit through a `delay_seconds` grace window, so a user banned
 * between the emit and the send is exactly the case this catches. The cost is one
 * primary-key lookup on a path that is about to open an SMTP conversation.
 *
 * **It still does not gate on `email_verified`, and Phase 9 kept it that way.**
 * §7.1 Q1's narrower half was settled by the org lead 2026-08-31: the gate
 * excludes an unverified user at ENQUEUE, in `emailChannel.eligible`, so no
 * outbox row is written and the admin reach preview can say how many were
 * dropped. Adding the same condition here as well would look like defence in
 * depth and would in fact be a second, invisible answer to the question — this
 * function is also what a password reset would reach if it ever routed through
 * the channel, and reset mail is deliberately ungated (`passwordReset.controller.js`).
 */
const addressFor = async (userId) => {
  const rows = await query(
    `SELECT email FROM users
      WHERE id = ? AND status = 'active' AND email IS NOT NULL AND email <> ''`,
    [Number(userId)],
  )
  return rows.length ? { address: rows[0].email } : null
}

/**
 * Which of these users hold an UNVERIFIED address - the set the Phase 1b gate
 * excludes when it is on (Phase 9).
 *
 * A user with no address at all is in this set, and that is not incidental: the
 * gate's question is "may we mail this person", and nowhere to send is a stronger
 * no than an unconfirmed somewhere. `addressFor` refuses them at delivery either
 * way; including them here is what stops an outbox row being written for a send
 * that is already known to be impossible.
 *
 * One query for a whole audience. The engine calls it once per rule per event
 * with up to MAX_AUDIENCE ids, so a per-user lookup would be five thousand round
 * trips on the path that is supposed to be the cheap one.
 */
const unverifiedAmong = async (userIds) => {
  const wanted = [...new Set(userIds.map(Number).filter((n) => Number.isInteger(n) && n > 0))]
  if (!wanted.length) return new Set()
  const capped = wanted.slice(0, MAX_AUDIENCE)
  const rows = await query(
    `SELECT id FROM users
      WHERE id IN (${marks(capped)})
        AND (email_verified = 0 OR email IS NULL OR email = '')`,
    capped,
  )
  return new Set(ids(rows))
}

/**
 * The addresses for a set of active users, as a Map - what the admin reach
 * preview hashes to count how many of them are suppressed.
 *
 * It returns PLAINTEXT, which is the one thing this subsystem otherwise avoids,
 * and there is no way around it: a suppression is keyed on the sha256 of an
 * address, so answering "how many of these people are suppressed" requires
 * hashing each one. The caller (`engagement.controller`) hashes immediately and
 * returns only a count - no route ever serializes what this returns.
 */
const addressesFor = async (userIds) => {
  const wanted = [...new Set(userIds.map(Number).filter((n) => Number.isInteger(n) && n > 0))]
  if (!wanted.length) return new Map()
  const capped = wanted.slice(0, MAX_AUDIENCE)
  const rows = await query(
    `SELECT id, email FROM users
      WHERE id IN (${marks(capped)}) AND status = 'active' AND email IS NOT NULL AND email <> ''`,
    capped,
  )
  return new Map(rows.map((r) => [Number(r.id), r.email]))
}

module.exports = {
  active,
  staff,
  subscribers,
  filterActive,
  storedModes,
  addressFor,
  unverifiedAmong,
  addressesFor,
  MAX_AUDIENCE,
}