remove ConsumerRecord

Author: acco

assets/svelte/components/FilterForm.svelte

@@ -3,7 +3,6 @@
   import { Switch } from "$lib/components/ui/switch";
   import FunctionPicker from "$lib/consumers/FunctionPicker.svelte";
 
-  export let messageKind: string;
   export let form: {
     actions: string[];
     filterId: string;
@@ -35,29 +34,27 @@
     <Label class="text-base font-medium">Filters</Label>
   {/if}
 
-  {#if messageKind === "event"}
-    <div class="flex flex-col gap-4">
-      <div class="text-base font-medium">Operations to capture</div>
-      <div class="flex items-center gap-4">
-        {#each switches as { id, label }}
-          <div class="flex items-center gap-2">
-            <Label for={id} class="cursor-pointer">{label}</Label>
-            <Switch
-              {id}
-              disabled={!form.postgresDatabaseId && !form.tableOid}
-              checked={actions.includes(id)}
-              onCheckedChange={(checked) => {
-                const newActions = checked
-                  ? [...actions, id]
-                  : actions.filter((a) => a !== id);
-                form.actions = newActions;
-              }}
-            />
-          </div>
-        {/each}
-      </div>
+  <div class="flex flex-col gap-4">
+    <div class="text-base font-medium">Operations to capture</div>
+    <div class="flex items-center gap-4">
+      {#each switches as { id, label }}
+        <div class="flex items-center gap-2">
+          <Label for={id} class="cursor-pointer">{label}</Label>
+          <Switch
+            {id}
+            disabled={!form.postgresDatabaseId && !form.tableOid}
+            checked={actions.includes(id)}
+            onCheckedChange={(checked) => {
+              const newActions = checked
+                ? [...actions, id]
+                : actions.filter((a) => a !== id);
+              form.actions = newActions;
+            }}
+          />
+        </div>
+      {/each}
     </div>
-  {/if}
+  </div>
 
   <!-- Filter Function -->
   <div class="flex flex-col gap-4">

assets/svelte/consumers/SinkConsumerForm.svelte

@@ -79,11 +79,8 @@
   };
   export let isSelfHosted: boolean;
 
-  type MessageKind = "event" | "record";
-
   interface FormState {
     type: string;
-    messageKind: MessageKind;
     messageGrouping: boolean;
     maxMemoryMb: number;
     postgresDatabaseId: string | null;
@@ -116,7 +113,6 @@
 
   let initialForm: FormState = {
     type: consumer.type,
-    messageKind: (consumer.message_kind || "event") as MessageKind,
     messageGrouping: consumer.message_grouping,
     maxMemoryMb: Number(consumer.max_memory_mb),
     postgresDatabaseId: consumer.postgres_database_id,
@@ -385,7 +381,6 @@
       <CardContent>
         <FilterForm
           {functions}
-          messageKind={form.messageKind}
           {selectedDatabase}
           bind:form
           {refreshFunctions}

assets/svelte/consumers/types.ts

@@ -29,7 +29,6 @@ export type BaseConsumer = {
   name: string;
   annotations: Record<string, boolean>;
   status: "active" | "paused" | "disabled";
-  message_kind: string;
   message_grouping: boolean;
   ack_wait_ms: number;
   max_ack_pending: number;

assets/svelte/databases/Show.svelte

@@ -29,7 +29,6 @@
   interface Consumer {
     id: string;
     name: string;
-    message_kind: string;
     consumer_kind: string;
     source_tables: Table[];
     href: string;

assets/svelte/sinks/nats/NatsSinkCard.svelte

@@ -91,12 +91,9 @@
                 class="underline">router</a
               >
               <ExternalLink class="h-4 w-4 inline" />
-            {:else if consumer.message_kind === "event"}
+            {:else}
               sequin.changes.{consumer.database
                 .name}.&lt;table_schema&gt;.&lt;table_name&gt;.&lt;action&gt;
-            {:else}
-              sequin.rows.{consumer.database
-                .name}.&lt;table_schema&gt;.&lt;table_name&gt;
             {/if}
           </span>
         </div>

assets/svelte/wal_pipelines/Form.svelte

@@ -157,7 +157,6 @@
 
           {#if selectedSourceTable}
             <FilterForm
-              messageKind="event"
               selectedTable={selectedSourceTable}
               bind:form
               onFilterChange={(filters) => (form.sourceTableFilters = filters)}

lib/sequin/changeset.ex

@@ -33,13 +33,7 @@ defmodule Sequin.Changeset do
   end
 
   def cast_embed(%Ecto.Changeset{} = changeset, :source_tables) do
-    case get_field(changeset, :message_kind) do
-      :record ->
-        cast_embed(changeset, :source_tables, with: &SourceTable.record_changeset(&1, &2))
-
-      :event ->
-        cast_embed(changeset, :source_tables, with: &SourceTable.event_changeset(&1, &2))
-    end
+    cast_embed(changeset, :source_tables, with: &SourceTable.event_changeset(&1, &2))
   end
 
   # See: https://github.com/sequinstream/sequin/issues/1465

lib/sequin/consumers/acknowledged_messages/acknowledged_messages.ex

@@ -3,7 +3,6 @@ defmodule Sequin.Consumers.AcknowledgedMessages do
 
   alias Sequin.Consumers.AcknowledgedMessages.AcknowledgedMessage
   alias Sequin.Consumers.ConsumerEvent
-  alias Sequin.Consumers.ConsumerRecord
   alias Sequin.Error
   alias Sequin.Redis
 
@@ -12,7 +11,7 @@ defmodule Sequin.Consumers.AcknowledgedMessages do
   @doc """
   Stores messages for a given consumer_id in a Redis sorted set and trims to the latest @max_messages.
   """
-  @spec store_messages(String.t(), list(ConsumerEvent.t() | ConsumerRecord.t()), non_neg_integer()) ::
+  @spec store_messages(String.t(), list(ConsumerEvent.t()), non_neg_integer()) ::
           :ok | {:error, Error.t()}
   def store_messages(consumer_id, messages, max_messages \\ @max_messages) do
     # Add messages to the sorted set
@@ -65,30 +64,6 @@ defmodule Sequin.Consumers.AcknowledgedMessages do
     "acknowledged_messages:v0:#{consumer_id}"
   end
 
-  def to_acknowledged_message(%ConsumerRecord{} = record) do
-    deliver_count = if record.deliver_count == 0, do: 1, else: record.deliver_count
-
-    %AcknowledgedMessage{
-      id: record.id,
-      consumer_id: record.consumer_id,
-      seq: record.commit_lsn + record.commit_idx,
-      commit_lsn: record.commit_lsn,
-      commit_idx: record.commit_idx,
-      commit_timestamp: record.data.metadata.commit_timestamp,
-      ack_id: record.ack_id,
-      deliver_count: deliver_count,
-      last_delivered_at: record.last_delivered_at,
-      record_pks: record.record_pks,
-      table_oid: record.table_oid,
-      not_visible_until: record.not_visible_until,
-      inserted_at: record.inserted_at,
-      trace_id: record.replication_message_trace_id,
-      state: record.state,
-      table_name: record.data.metadata.table_name,
-      table_schema: record.data.metadata.table_schema
-    }
-  end
-
   def to_acknowledged_message(%ConsumerEvent{} = event) do
     deliver_count = if event.deliver_count == 0, do: 1, else: event.deliver_count