Add document_mode option and AUTH_DISABLED env var for self-hosted

.dockerignore

@@ -0,0 +1,72 @@
+# 2026-05-11: Upgraded from blacklist to *whitelist*-style. The Dockerfile
+# (packages/sequin/Dockerfile) COPYs a very specific set of paths — every
+# other path in this directory ships across the SSH tunnel to Hetzner
+# on every deploy for zero benefit. Audit before the change:
+#
+#   docs/         — 28MB of mostly markdown (only docs/snippets/
+#                   function-transform-snippet.mdx is actually COPYd)
+#   test/         — 1.5MB tests, never built into the prod image
+#   website/      — 216KB marketing site
+#   examples/     — 376KB usage examples
+#   cli/          — 384KB Go code; the Dockerfile builds the CLI from
+#                   `git clone` in a separate `cli-builder` stage, our
+#                   local copy is dead weight
+#   deployment/   — 268KB k8s/helm templates
+#   *.md root     — 12-32KB each, none referenced by build
+#
+# Stripping these cuts the build context from 39MB → ~9MB. Smaller
+# context means less time spent sending bytes over SSH, fewer files for
+# BuildKit to hash, and fewer paths where stat/mtime jitter can trick
+# the context hash into "different" on otherwise-identical content.
+#
+# Original 2026-04-28 fix (sst-env.d.ts + blacklist) is preserved
+# inside the explicit exclusions below.
+
+# Exclude everything by default — whitelist below allows back what
+# Dockerfile actually uses
+*
+
+# ─── Whitelist: paths the Dockerfile COPYs ────────────────────────
+!.iex.exs
+!Dockerfile
+!mix.exs
+!mix.lock
+!config
+!lib
+!priv
+!assets
+!rel
+!scripts
+!docs
+
+# Subtree restrictions inside whitelisted dirs — exclude all then
+# allow back the single needed file.
+docs/*
+!docs/snippets
+docs/snippets/*
+!docs/snippets/function-transform-snippet.mdx
+scripts/*
+!scripts/start_commands.sh
+
+# ─── Re-blacklist common cruft under whitelisted dirs ─────────────
+# These supersede whitelist for transient / regenerated / dev-only files.
+**/sst-env.d.ts
+**/node_modules
+**/_build
+**/deps
+**/.elixir_ls
+**/cover
+**/erl_crash.dump
+**/.git
+**/.gitignore
+**/.gitkeep
+**/.DS_Store
+**/*.log
+**/*.tmp
+**/tmp
+
+# Editor / IDE
+**/.vscode
+**/.idea
+**/.aider*
+**/CLAUDE.md

.gitignore

@@ -72,3 +72,6 @@ test-config/
 typesense-data/
 meilisearch-data/
 .expert/
+
+# SST auto-generated type stubs (from monorepo builds that scan package.json dirs)
+sst-env.d.ts

Dockerfile

@@ -1,16 +1,24 @@
+# ADR-0059: built in GitHub Actions, consumed by digest. This image builds with
+# PACKAGE-DIR context (context: packages/sequin, set via the build workflow's
+# per-image matrix) — NOT repo-root — because (a) the COPYs below are
+# package-relative, and (b) packages/sequin/.dockerignore is a whitelist that
+# keeps the one needed docs/snippets file, whereas the repo-root .dockerignore
+# excludes packages/sequin/docs entirely. SELF_HOSTED defaults to 1 (this repo
+# always self-hosts; it was the only build-arg the old dockerbuild.Image passed)
+# so no workflow build-arg is needed.
 ARG ELIXIR_VERSION=1.19.4
 ARG OTP_VERSION=28.3
 ARG DEBIAN_VERSION=trixie-20251208-slim
 ARG RELEASE_VERSION
 
-ARG BUILDER_IMAGE="hexpm/elixir:${ELIXIR_VERSION}-erlang-${OTP_VERSION}-debian-${DEBIAN_VERSION}"
+ARG BUILDER_IMAGE="hexpm/elixir:${ELIXIR_VERSION}-erlang-${OTP_VERSION}-debian-${DEBIAN_VERSION}@sha256:68e6d26c37a8f86b5ec6294cbbf034b9cab5268c7b4179f585f58a7b9c32e825"
 # Ensure Elixir installed on the runtime image for tooling purposes
-ARG RUNNER_IMAGE="hexpm/elixir:${ELIXIR_VERSION}-erlang-${OTP_VERSION}-debian-${DEBIAN_VERSION}"
+ARG RUNNER_IMAGE="hexpm/elixir:${ELIXIR_VERSION}-erlang-${OTP_VERSION}-debian-${DEBIAN_VERSION}@sha256:68e6d26c37a8f86b5ec6294cbbf034b9cab5268c7b4179f585f58a7b9c32e825"
 
-ARG SELF_HOSTED=0
+ARG SELF_HOSTED=1
 
 # ---- CLI Build Stage ----
-FROM golang:1.24.2-bullseye AS cli-builder
+FROM golang:1.24.2-bullseye@sha256:f50ff25f8331682b44c1582974eb9e620fcb08052fc6ed434f93ca24636fc4d6 AS cli-builder
 
 ARG SEQUIN_CLI_VERSION=0.13.1
 RUN git clone --depth 1 --branch v${SEQUIN_CLI_VERSION} https://github.com/sequinstream/sequin.git /tmp/sequin \
@@ -86,8 +94,8 @@ ENV RELEASE_VERSION=${RELEASE_VERSION}
 # Compile the release
 RUN mix compile
 
-# Ensure stacktraces we send to Sentry are complete
-RUN mix sentry.package_source_code
+# Ensure stacktraces we send to Sentry are complete (skip for self-hosted — no DSN)
+RUN if [ "$SELF_HOSTED" != "1" ]; then mix sentry.package_source_code; fi
 
 # Changes to config/runtime.exs don't require recompiling the code
 COPY config/runtime.exs config/

config/prod.exs

@@ -8,8 +8,14 @@ import Config
 
 self_hosted = System.get_env("SELF_HOSTED", "0") in ~w(1 true)
 
+# Empty string DSN crashes Sentry — treat "" as nil for self-hosted builds
+sentry_dsn = case System.get_env("SENTRY_DSN") do
+  "" -> nil
+  dsn -> dsn
+end
+
 config :sentry,
-  dsn: System.get_env("SENTRY_DSN"),
+  dsn: sentry_dsn,
   release: System.get_env("RELEASE_VERSION")
 
 config :sequin, Sequin.ConsoleLogger, drop_metadata_keys: [:mfa]

lib/sequin/accounts/accounts.ex

@@ -97,6 +97,16 @@ defmodule Sequin.Accounts do
   """
   def get_user!(id), do: Repo.get!(User, id)
 
+  def get_first_user do
+    User
+    |> Ecto.Query.first()
+    |> Repo.one()
+    |> case do
+      nil -> nil
+      user -> Repo.preload(user, [:accounts, :accounts_users])
+    end
+  end
+
   def get_user_with_preloads!(user_id) do
     User
     |> Repo.get!(user_id)

lib/sequin/consumers/meilisearch_sink.ex

@@ -5,7 +5,7 @@ defmodule Sequin.Consumers.MeilisearchSink do
 
   import Ecto.Changeset
 
-  @derive {Jason.Encoder, only: [:endpoint_url, :index_name, :primary_key]}
+  @derive {Jason.Encoder, only: [:endpoint_url, :index_name, :primary_key, :document_mode]}
   @derive {Inspect, except: [:api_key]}
 
   @primary_key false
@@ -18,6 +18,7 @@ defmodule Sequin.Consumers.MeilisearchSink do
     field(:batch_size, :integer, default: 100)
     field(:timeout_seconds, :integer, default: 5)
     field(:routing_mode, Ecto.Enum, values: [:dynamic, :static])
+    field(:document_mode, Ecto.Enum, values: [:replace, :update], default: :replace)
   end
 
   def changeset(struct, params) do
@@ -29,7 +30,8 @@ defmodule Sequin.Consumers.MeilisearchSink do
       :api_key,
       :batch_size,
       :timeout_seconds,
-      :routing_mode
+      :routing_mode,
+      :document_mode
     ])
     |> validate_required([:endpoint_url, :api_key])
     |> validate_routing()

lib/sequin/metrics/prometheus.ex

@@ -156,6 +156,12 @@ defmodule Sequin.Prometheus do
       labels: [:consumer_id, :consumer_name]
     )
 
+    Counter.declare(
+      name: :sequin_message_discard_count,
+      help: "Total messages permanently discarded (dead-lettered) after exhausting max_retry_count.",
+      labels: [:consumer_id, :consumer_name]
+    )
+
     Gauge.new(
       name: :sequin_messages_in_delivery,
       labels: [:consumer_id, :consumer_name],
@@ -321,6 +327,12 @@ defmodule Sequin.Prometheus do
     Counter.inc([name: :sequin_message_deliver_failure_count, labels: [consumer_id, consumer_name]], count)
   end
 
+  @spec increment_message_discard_count(consumer_id :: String.t(), consumer_name :: String.t(), count :: number()) ::
+          :ok
+  def increment_message_discard_count(consumer_id, consumer_name, count \\ 1) do
+    Counter.inc([name: :sequin_message_discard_count, labels: [consumer_id, consumer_name]], count)
+  end
+
   @spec observe_messages_ingested_latency(consumer_id :: String.t(), consumer_name :: String.t(), latency_ms :: number()) ::
           :ok
   def observe_messages_ingested_latency(consumer_id, consumer_name, latency_ms) do

lib/sequin/runtime/slot_message_store.ex

@@ -762,6 +762,16 @@ defmodule Sequin.Runtime.SlotMessageStore do
     Sequin.ProcessMetrics.increment_throughput("messages_discarded", length(discarded_messages))
     Sequin.ProcessMetrics.gauge("message_count", map_size(state.messages))
 
+    if length(discarded_messages) > 0 do
+      Prometheus.increment_message_discard_count(state.consumer.id, state.consumer.name, length(discarded_messages))
+
+      Logger.warning(
+        "[SlotMessageStore] Permanently discarding #{length(discarded_messages)} message(s) after exhausting max_retry_count (consumer=#{state.consumer.name})",
+        consumer_id: state.consumer.id,
+        partition: state.partition
+      )
+    end
+
     with :ok <- handle_discarded_messages(state, discarded_messages),
          :ok <- upsert_messages(state, messages) do
       state = State.put_persisted_messages(state, messages)

lib/sequin/sinks/meilisearch/client.ex

@@ -19,79 +19,91 @@ defmodule Sequin.Sinks.Meilisearch.Client do
 
   defp decode_body(_), do: %{}
 
+  # Polls Meilisearch's /tasks/{id} endpoint until the task reaches a terminal
+  # state (succeeded/failed) or the deadline derived from sink.timeout_seconds
+  # elapses. Replaces the upstream Req-retry-based loop, whose hardcoded
+  # 5-retry budget (~3-6s) was way under our auctions index's normal
+  # enqueued→finished latency (15-20s under load), causing every batch to
+  # appear to fail even though Meilisearch was indexing fine.
+  # Cherry-picked from sequinstream/sequin#2140 (PierreMesure).
   defp wait_for_task(%MeilisearchSink{} = sink, task_id) do
-    req =
-      sink
-      |> base_request()
-      |> Req.merge(
-        url: "/tasks/#{task_id}",
-        # We need to disable this if we use a custom retry that emits delays
-        retry_delay: nil,
-        max_retries: 5,
-        retry: fn request, response_or_exception ->
-          should_retry =
-            case response_or_exception do
-              %Req.Response{status: 200, body: encoded_body} ->
-                # NOTE: Req does not automatically decode on retry functions
-                case encoded_body |> :zlib.gunzip() |> Jason.decode() do
-                  {:ok, %{"status" => status}} when status in ["enqueued", "processing"] ->
-                    true
-
-                  _ ->
-                    false
-                end
-
-              %Req.Response{status: status} when status in [408, 429, 500, 502, 503, 504] ->
-                true
-
-              %Req.TransportError{reason: reason} when reason in [:timeout, :econnrefused, :closed] ->
-                true
-
-              _ ->
-                false
-            end
-
-          if should_retry do
-            # Req tracks retry count internally via request.private[:req_retry_count]
-            count = request.private[:req_retry_count] || 0
-            delay = Sequin.Time.exponential_backoff(200, count, 10_000)
-
-            Logger.debug("[Meilisearch] Task #{task_id} has not succeeded, retrying in #{delay}ms (attempt #{count + 1})")
-
-            {:delay, delay}
-          else
-            false
+    deadline_ms = System.monotonic_time(:millisecond) + sink.timeout_seconds * 1000
+    do_wait_for_task(sink, task_id, deadline_ms, 0, nil)
+  end
+
+  defp do_wait_for_task(%MeilisearchSink{} = sink, task_id, deadline_ms, attempt, last_status) do
+    now_ms = System.monotonic_time(:millisecond)
+    remaining_ms = deadline_ms - now_ms
+
+    if remaining_ms <= 0 do
+      {:error,
+       Error.service(
+         service: :meilisearch,
+         message: "Task verification timed out",
+         details: %{task_id: task_id, last_status: last_status}
+       )}
+    else
+      req =
+        sink
+        |> base_request()
+        |> Req.merge(
+          url: "/tasks/#{task_id}",
+          receive_timeout: min(1_000, remaining_ms),
+          retry: false,
+          max_retries: 0
+        )
+
+      case Req.get(req) do
+        {:ok, %{status: 200, body: body}} ->
+          decoded_body = decode_body(body)
+
+          case decoded_body do
+            %{"status" => status} when status in ["succeeded", "success"] ->
+              :ok
+
+            %{"status" => "failed", "error" => error} ->
+              message = extract_error_message(error)
+              {:error, Error.service(service: :meilisearch, message: message, details: error)}
+
+            %{"status" => status} when status in ["enqueued", "processing"] ->
+              retry_task_poll(sink, task_id, deadline_ms, attempt, status)
+
+            _ ->
+              {:error, Error.service(service: :meilisearch, message: "Invalid response format")}
           end
-        end
-      )
 
-    case Req.get(req) do
-      {:ok, %{body: body}} ->
-        decoded_body = decode_body(body)
+        {:error, %Req.TransportError{reason: reason}}
+        when reason in [:timeout, :econnrefused, :closed] ->
+          retry_task_poll(sink, task_id, deadline_ms, attempt, last_status)
 
-        case decoded_body do
-          %{"status" => status} when status in ["succeeded", "success"] ->
-            :ok
+        {:ok, %{status: status}} when status in [408, 429, 500, 502, 503, 504] ->
+          Logger.debug("[Meilisearch] Task #{task_id} check returned #{status}, retrying")
+          retry_task_poll(sink, task_id, deadline_ms, attempt, last_status)
 
-          %{"status" => "failed", "error" => error} ->
-            message = extract_error_message(error)
-            {:error, Error.service(service: :meilisearch, message: message, details: error)}
+        {:error, reason} ->
+          {:error, Error.service(service: :meilisearch, message: "Unknown error", details: reason)}
+      end
+    end
+  end
 
-          %{"status" => status} when status in ["enqueued", "processing"] ->
-            # This means we exhausted retries
-            {:error,
-             Error.service(
-               service: :meilisearch,
-               message: "Task verification timed out",
-               details: %{task_id: task_id, last_status: status}
-             )}
+  defp retry_task_poll(%MeilisearchSink{} = sink, task_id, deadline_ms, attempt, last_status) do
+    now_ms = System.monotonic_time(:millisecond)
+    remaining_ms = deadline_ms - now_ms
 
-          _ ->
-            {:error, Error.service(service: :meilisearch, message: "Invalid response format")}
-        end
+    if remaining_ms <= 0 do
+      {:error,
+       Error.service(
+         service: :meilisearch,
+         message: "Task verification timed out",
+         details: %{task_id: task_id, last_status: last_status}
+       )}
+    else
+      delay = min(Sequin.Time.exponential_backoff(200, attempt, 10_000), remaining_ms)
 
-      {:error, reason} ->
-        {:error, Error.service(service: :meilisearch, message: "Unknown error", details: reason)}
+      Logger.debug("[Meilisearch] Task #{task_id} has not succeeded, retrying in #{delay}ms (attempt #{attempt + 1})")
+
+      Process.sleep(delay)
+      do_wait_for_task(sink, task_id, deadline_ms, attempt + 1, last_status)
     end
   end
 
@@ -110,7 +122,15 @@ defmodule Sequin.Sinks.Meilisearch.Client do
         body: jsonl
       )
 
-    case Req.post(req) do
+    # :replace (default) uses POST = add or replace (full document replacement)
+    # :update uses PUT = add or update (partial merge — only overwrites fields present)
+    result =
+      case sink.document_mode do
+        :update -> Req.put(req)
+        _ -> Req.post(req)
+      end
+
+    case result do
       {:ok, %{body: %{"taskUid" => task_id}}} ->
         wait_for_task(sink, task_id)