diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 000000000..3a55300bf --- /dev/null +++ b/.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 diff --git a/.gitignore b/.gitignore index 14e883b19..0839516bd 100644 --- a/.gitignore +++ b/.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 diff --git a/Dockerfile b/Dockerfile index 7040d9c34..47f442c9c 100644 --- a/Dockerfile +++ b/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 \ @@ -18,7 +26,7 @@ RUN git clone --depth 1 --branch v${SEQUIN_CLI_VERSION} https://github.com/sequi && go build -o /sequin-cli # ---- Elixir Build Stage ---- -FROM ${BUILDER_IMAGE} AS builder +FROM ${BUILDER_IMAGE} AS deps # Pass the SELF_HOSTED arg as an environment variable ARG SELF_HOSTED @@ -59,6 +67,15 @@ RUN mkdir config COPY config/config.exs config/${MIX_ENV}.exs config/ RUN mix deps.compile +# ---- stage boundary (CI cold-build fix, 2026-07-19) ---- +# `deps` above is the heavy, rarely-invalidated half (toolchain + node + hex +# deps compile). CI builds and registry-caches it ALONE: a single-shot cold +# build's scratch breaches the build node's disk-eviction floor, but each half +# fits. `builder` inherits deps' ENV (MIX_ENV, LANG, ERL_FLAGS, SELF_HOSTED, +# SENTRY_DSN) automatically; ARG does not inherit, but none is read below +# except RELEASE_VERSION, which is re-declared where it was already. +FROM deps AS builder + COPY priv priv COPY lib lib @@ -86,8 +103,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/ diff --git a/config/prod.exs b/config/prod.exs index e9ce90920..e17a0acca 100644 --- a/config/prod.exs +++ b/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] diff --git a/lib/sequin/accounts/accounts.ex b/lib/sequin/accounts/accounts.ex index 26ce7f585..037823f08 100644 --- a/lib/sequin/accounts/accounts.ex +++ b/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) diff --git a/lib/sequin/consumers/meilisearch_sink.ex b/lib/sequin/consumers/meilisearch_sink.ex index b7a926052..9f3f6045a 100644 --- a/lib/sequin/consumers/meilisearch_sink.ex +++ b/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() diff --git a/lib/sequin/metrics/prometheus.ex b/lib/sequin/metrics/prometheus.ex index 45159dbb8..e5eb2b48e 100644 --- a/lib/sequin/metrics/prometheus.ex +++ b/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 diff --git a/lib/sequin/runtime/slot_message_store.ex b/lib/sequin/runtime/slot_message_store.ex index 838caf8c1..2417f63cc 100644 --- a/lib/sequin/runtime/slot_message_store.ex +++ b/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) diff --git a/lib/sequin/sinks/meilisearch/client.ex b/lib/sequin/sinks/meilisearch/client.ex index 26effd614..0b1166fca 100644 --- a/lib/sequin/sinks/meilisearch/client.ex +++ b/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) diff --git a/lib/sequin/transforms/transforms.ex b/lib/sequin/transforms/transforms.ex index ebd12dffc..6c6b9c1aa 100644 --- a/lib/sequin/transforms/transforms.ex +++ b/lib/sequin/transforms/transforms.ex @@ -413,7 +413,8 @@ defmodule Sequin.Transforms do index_name: sink.index_name, primary_key: sink.primary_key, api_key: SensitiveValue.new(sink.api_key, show_sensitive), - timeout_seconds: sink.timeout_seconds + timeout_seconds: sink.timeout_seconds, + document_mode: if(sink.document_mode != :replace, do: to_string(sink.document_mode)) }) end @@ -860,6 +861,29 @@ defmodule Sequin.Transforms do {:halt, {:error, Error.validation(summary: "batch_size must be a positive integer <= 1000")}} end + "batch_timeout_ms" -> + case value do + ms when is_integer(ms) and ms > 0 -> + {:cont, {:ok, Map.put(acc, :batch_timeout_ms, ms)}} + + _ -> + {:halt, {:error, Error.validation(summary: "batch_timeout_ms must be a positive integer")}} + end + + # Per-sink memory cap, exposed in catchseo fork so the YAML can + # bump it for high-volume sinks (e.g. meilisearch-auctions during + # daily inventory-sync flood). Falls back to schema default (512 + # MB in catchseo fork) when omitted. Validation matches the + # schema's `greater_than_or_equal_to: 128` rule. + "max_memory_mb" -> + case value do + mb when is_integer(mb) and mb >= 128 -> + {:cont, {:ok, Map.put(acc, :max_memory_mb, mb)}} + + _ -> + {:halt, {:error, Error.validation(summary: "max_memory_mb must be an integer >= 128")}} + end + "actions" -> {:cont, {:ok, Map.put(acc, :actions, value)}} @@ -1305,7 +1329,8 @@ defmodule Sequin.Transforms do primary_key: attrs["primary_key"], api_key: attrs["api_key"], batch_size: attrs["batch_size"], - timeout_seconds: attrs["timeout_seconds"] + timeout_seconds: attrs["timeout_seconds"], + document_mode: attrs["document_mode"] }} end diff --git a/lib/sequin/yaml_loader.ex b/lib/sequin/yaml_loader.ex index 55a105e74..52e9944ae 100644 --- a/lib/sequin/yaml_loader.ex +++ b/lib/sequin/yaml_loader.ex @@ -372,6 +372,25 @@ defmodule Sequin.YamlLoader do database = Sequin.Repo.preload(database, :replication_slot) + # Re-introspect the live table list before sink validation. The stored + # struct's `.tables` cache is a snapshot from the last refresh; a config + # that references a table created since (e.g. by a migration in the same + # deploy) would otherwise fail find_table/2 at boot forever — the apply + # crashes before any background refresh can run (crash-loop observed + # live 2026-07-18 adding `ai_classification`). + database = + case Databases.update_tables(database) do + {:ok, refreshed} -> + refreshed + + {:error, error} -> + Logger.warning( + "Could not refresh tables for database '#{database.name}' before config apply: #{inspect(error)}" + ) + + database + end + with {:ok, database_attrs} <- parse_database_update_attrs(db_params, database.replication_slot) do update_database(database, database_attrs) end diff --git a/lib/sequin_web/user_auth.ex b/lib/sequin_web/user_auth.ex index 2f980b44b..38cb50997 100644 --- a/lib/sequin_web/user_auth.ex +++ b/lib/sequin_web/user_auth.ex @@ -206,12 +206,19 @@ defmodule SequinWeb.UserAuth do if socket.assigns.current_user do {:cont, socket} else - socket = - socket - |> Phoenix.LiveView.put_flash(:toast, %{kind: :error, title: "Please log in to continue."}) - |> Phoenix.LiveView.redirect(to: ~p"/login") + if auth_disabled?() do + case Accounts.get_first_user() do + nil -> {:cont, socket} + user -> {:cont, Phoenix.Component.assign(socket, :current_user, user)} + end + else + socket = + socket + |> Phoenix.LiveView.put_flash(:toast, %{kind: :error, title: "Please log in to continue."}) + |> Phoenix.LiveView.redirect(to: ~p"/login") - {:halt, socket} + {:halt, socket} + end end end @@ -279,20 +286,24 @@ defmodule SequinWeb.UserAuth do if conn.assigns[:current_user] do conn else - {title, redirect_to} = - case Keyword.get(opts, :unauthenticated_redirect, :login) do - :login -> - {"Please log in to continue.", ~p"/login"} + if auth_disabled?() do + auto_login_default_user(conn) + else + {title, redirect_to} = + case Keyword.get(opts, :unauthenticated_redirect, :login) do + :login -> + {"Please log in to continue.", ~p"/login"} - :register -> - {"Please register to continue.", ~p"/register"} - end + :register -> + {"Please register to continue.", ~p"/register"} + end - conn - |> put_flash(:toast, %{kind: :error, title: title}) - |> maybe_store_return_to() - |> redirect(to: redirect_to) - |> halt() + conn + |> put_flash(:toast, %{kind: :error, title: title}) + |> maybe_store_return_to() + |> redirect(to: redirect_to) + |> halt() + end end end @@ -309,4 +320,23 @@ defmodule SequinWeb.UserAuth do defp maybe_store_return_to(conn), do: conn defp signed_in_path(_conn), do: ~p"/" + + defp auth_disabled? do + System.get_env("AUTH_DISABLED") in ~w(true 1) + end + + defp auto_login_default_user(conn) do + case Accounts.get_first_user() do + nil -> + conn + + user -> + token = Accounts.generate_user_session_token(user) + + conn + |> renew_session() + |> put_token_in_session(token) + |> assign(:current_user, user) + end + end end diff --git a/test/sequin/meilisearch_client_test.exs b/test/sequin/meilisearch_client_test.exs index d6441c692..f506fed80 100644 --- a/test/sequin/meilisearch_client_test.exs +++ b/test/sequin/meilisearch_client_test.exs @@ -67,6 +67,59 @@ defmodule Sequin.Sinks.Meilisearch.ClientTest do end end + describe "import_documents/3 with document_mode: :update" do + @update_sink %MeilisearchSink{ + type: :meilisearch, + endpoint_url: "http://127.0.0.1:7700", + index_name: "test", + primary_key: "id", + api_key: "token", + document_mode: :update + } + + test "uses PUT when document_mode is :update" do + records = [SinkFactory.meilisearch_record()] + + Req.Test.expect(Client, fn conn -> + assert conn.method == "PUT" + assert conn.request_path == "/indexes/test/documents" + + Req.Test.json(conn, %{"taskUid" => 1}) + end) + + Req.Test.expect(Client, fn conn -> + assert conn.method == "GET" + assert conn.request_path == "/tasks/1" + + response_data = %{"status" => "succeeded"} + send_gzipped_response(conn, 200, response_data) + end) + + assert :ok = Client.import_documents(@update_sink, "test", records) + end + + test "uses POST when document_mode is :replace (default)" do + records = [SinkFactory.meilisearch_record()] + + Req.Test.expect(Client, fn conn -> + assert conn.method == "POST" + assert conn.request_path == "/indexes/test/documents" + + Req.Test.json(conn, %{"taskUid" => 1}) + end) + + Req.Test.expect(Client, fn conn -> + assert conn.method == "GET" + assert conn.request_path == "/tasks/1" + + response_data = %{"status" => "succeeded"} + send_gzipped_response(conn, 200, response_data) + end) + + assert :ok = Client.import_documents(@sink, "test", records) + end + end + describe "delete_documents/2" do test "successfully delete batch" do records = [SinkFactory.meilisearch_record(), SinkFactory.meilisearch_record()] @@ -305,7 +358,7 @@ defmodule Sequin.Sinks.Meilisearch.ClientTest do assert_receive {:task_check, 3}, 2000 end - test "wait_for_task times out after exhausting retries" do + test "wait_for_task times out after exhausting the timeout budget" do test_pid = self() call_count = :counters.new(1, []) records = [SinkFactory.meilisearch_record()] @@ -339,7 +392,7 @@ defmodule Sequin.Sinks.Meilisearch.ClientTest do assert error.details.task_id == 456 assert error.details.last_status == "processing" - # Verify we made all 6 attempts (1 initial + 5 retries) + # Verify we polled until the total timeout budget was exhausted. for i <- 0..5 do assert_receive {:task_check, ^i}, 3_000 end diff --git a/test/sequin/meilisearch_sink_test.exs b/test/sequin/meilisearch_sink_test.exs index 6f797d602..c1b280aae 100644 --- a/test/sequin/meilisearch_sink_test.exs +++ b/test/sequin/meilisearch_sink_test.exs @@ -55,5 +55,21 @@ defmodule Sequin.Consumers.MeilisearchSinkTest do refute :index_name in changeset.changes end + + test "defaults document_mode to :replace", %{valid_params: params} do + changeset = MeilisearchSink.changeset(%MeilisearchSink{}, params) + assert Ecto.Changeset.get_field(changeset, :document_mode) == :replace + end + + test "accepts document_mode :update", %{valid_params: params} do + changeset = MeilisearchSink.changeset(%MeilisearchSink{}, Map.put(params, :document_mode, :update)) + assert Sequin.Error.errors_on(changeset) == %{} + assert Ecto.Changeset.get_field(changeset, :document_mode) == :update + end + + test "rejects invalid document_mode", %{valid_params: params} do + changeset = MeilisearchSink.changeset(%MeilisearchSink{}, Map.put(params, :document_mode, :invalid)) + assert Sequin.Error.errors_on(changeset)[:document_mode] != nil + end end end