PhoenixKitSync.ConnectionNotifier (PhoenixKitSync v0.3.0)

Copy Markdown View Source

Handles cross-site notification when creating sender connections.

When a sender connection is created, this module notifies the remote site so they can automatically register the incoming connection on their end.

How It Works

  1. When you create a sender connection pointing to a remote site (e.g., "https://remote.com")
  2. This module calls POST https://remote.com{prefix}/sync/api/register-connection where {prefix} mirrors the remote's configured URL prefix (see remote_url_prefix/0)
  3. The remote site creates a receiver connection automatically
  4. The result is recorded in the connection's metadata

Remote Site Responses

  • 200 OK - Connection registered successfully
  • 401 Unauthorized - Password required or invalid
  • 403 Forbidden - Incoming connections denied
  • 409 Conflict - Connection already exists
  • 503 Service Unavailable - DB Sync module disabled

Usage

Usually called automatically when creating connections via the LiveView UI. Can also be called manually:

{:ok, result} = ConnectionNotifier.notify_remote_site(connection, token, password: "optional")

Summary

Types

A connection as these functions read it: the Connection struct, or any map with site_url and auth_token_hash under atom or string keys.

What a pull imported: record counts, unknown_columns when records carried columns the local table lacks, and truncated: true when the sender stopped short and the pull could not continue, with unsupported_key_type when the sender named the key type it has no cursor for.

Sender key to local key, per table: %{{table, sender_key} => local_key}.

Functions

Checks the status of a remote site's DB Sync API.

Fetches the list of available tables from the sender.

Fetch table records from a sender site via HTTP API for preview.

Fetch table schema from a sender site via HTTP API.

Resolves this site's own URL from Settings, config, or dynamic detection. Used to identify ourselves when communicating with remote sites.

Notifies a remote site to delete a connection.

Notifies a remote site about a new sender connection.

Notifies a remote site of a status change (suspend, reactivate, revoke).

Pulls data for a specific table from the sender.

Same as pull_table_data but accepts and returns a uuid_remap for FK remapping across tables. Returns {:ok, import_result, updated_remap} or {:error, reason, unchanged_remap}, with the same error reasons as pull_table_data/3.

Queries the sender for the current connection status.

Returns the URL prefix the notifier prepends to remote sync API paths.

Verifies a connection still exists on the remote site.

Types

connection_ref()

@type connection_ref() :: PhoenixKitSync.Connection.t() | map()

A connection as these functions read it: the Connection struct, or any map with site_url and auth_token_hash under atom or string keys.

notify_result()

@type notify_result() :: %{
  success: boolean(),
  status: :registered | :pending | :failed | :skipped,
  message: String.t(),
  remote_connection_uuid: String.t() | nil,
  http_status: integer() | nil,
  error: String.t() | nil
}

pull_result()

@type pull_result() :: %{
  :imported => non_neg_integer(),
  :skipped => non_neg_integer(),
  :errors => non_neg_integer(),
  optional(:unknown_columns) => [String.t()],
  optional(:truncated) => true,
  optional(:unsupported_key_type) => String.t()
}

What a pull imported: record counts, unknown_columns when records carried columns the local table lacks, and truncated: true when the sender stopped short and the pull could not continue, with unsupported_key_type when the sender named the key type it has no cursor for.

uuid_remap()

@type uuid_remap() :: %{optional({String.t(), String.t()}) => term()}

Sender key to local key, per table: %{{table, sender_key} => local_key}.

Functions

check_remote_status(site_url)

@spec check_remote_status(String.t()) :: {:ok, map()} | {:error, any()}

Checks the status of a remote site's DB Sync API.

Parameters

  • site_url - The remote site's base URL

Returns

  • {:ok, status} - Remote site status
  • {:error, reason} - Failed to contact site

fetch_sender_tables(connection, opts \\ [])

@spec fetch_sender_tables(connection_ref(), keyword()) ::
  {:ok, [map()]} | {:error, term()}

Fetches the list of available tables from the sender.

Called by receiver to get a list of tables that can be synced.

Parameters

  • connection - The receiver connection (must have site_url and auth_token/auth_token_hash)
  • opts - Options:
    • :timeout - HTTP request timeout (default: 30_000ms)

Returns

  • {:ok, tables} - List of table info maps with :name, :row_count, :size_bytes
  • {:error, :offline} - Sender is offline
  • {:error, reason} - Failed to fetch

fetch_table_records(connection, table_name, opts \\ [])

@spec fetch_table_records(connection_ref(), String.t(), keyword()) ::
  {:ok, [map()]} | {:error, term()}

Fetch table records from a sender site via HTTP API for preview.

Options:

  • :limit - Maximum number of records to fetch (default: 10)
  • :offset - Offset for pagination (default: 0)
  • :ids - List of specific IDs to fetch
  • :id_range - Tuple of {start_id, end_id}

Returns:

  • {:ok, records} - List of record maps
  • {:error, :offline} - Sender is offline
  • {:error, reason} - Failed to fetch records

fetch_table_schema(connection, table_name, opts \\ [])

@spec fetch_table_schema(connection_ref(), String.t(), keyword()) ::
  {:ok, map()} | {:error, term()}

Fetch table schema from a sender site via HTTP API.

Returns:

  • {:ok, schema} - Map with :columns list
  • {:error, :offline} - Sender is offline
  • {:error, reason} - Failed to fetch schema

get_our_site_url()

@spec get_our_site_url() :: String.t()

Resolves this site's own URL from Settings, config, or dynamic detection. Used to identify ourselves when communicating with remote sites.

notify_delete(connection, opts \\ [])

@spec notify_delete(connection_ref(), keyword()) ::
  {:ok, :deleted | :not_found | :offline} | {:error, term()}

Notifies a remote site to delete a connection.

Called when a receiver deletes their connection - notifies the sender to also delete.

Parameters

  • connection - The connection being deleted (must have site_url and auth_token_hash)
  • opts - Options:
    • :timeout - HTTP request timeout (default: 30_000ms)

Returns

  • {:ok, :deleted} - Remote site deleted the connection
  • {:ok, :not_found} - Connection didn't exist on remote (already deleted)
  • {:ok, :offline} - Remote site is offline (will self-heal later)
  • {:error, reason} - Failed to notify

notify_remote_site(connection, raw_token, opts \\ [])

@spec notify_remote_site(map(), String.t(), keyword()) ::
  {:ok, notify_result()} | {:error, any()}

Notifies a remote site about a new sender connection.

Parameters

  • connection - The sender connection that was just created
  • raw_token - The raw auth token (only available at creation time)
  • opts - Options:
    • :password - Password to provide to remote site (if required)
    • :timeout - HTTP request timeout (default: 30_000ms)

Returns

  • {:ok, result} - Notification sent, result contains details
  • {:error, reason} - Failed to send notification

notify_status_change(connection, new_status, opts \\ [])

@spec notify_status_change(connection_ref(), String.t(), keyword()) ::
  {:ok, :updated | :skipped | :not_found | :offline} | {:error, term()}

Notifies a remote site of a status change (suspend, reactivate, revoke).

Called when a sender changes their connection status - the receiver should mirror it.

Parameters

  • connection - The connection with updated status
  • new_status - The new status ("active", "suspended", "revoked")
  • opts - Options:
    • :timeout - HTTP request timeout (default: 30_000ms)

Returns

  • {:ok, :updated} - Remote site updated the status
  • {:ok, :skipped} - A "receiver" connection: nothing is sent
  • {:ok, :not_found} - Connection not found on remote
  • {:ok, :offline} - Remote site is offline
  • {:error, reason} - Failed to notify

pull_table_data(connection, table_name, opts \\ [])

@spec pull_table_data(connection_ref(), String.t(), keyword()) ::
  {:ok, pull_result()} | {:error, term()}

Pulls data for a specific table from the sender.

Called by receiver to fetch table data during sync.

Parameters

  • connection - The receiver connection
  • table_name - Name of the table to pull
  • opts - Options:
    • :timeout - HTTP request timeout (default: 60_000ms for large data)
    • :conflict_strategy - How to handle existing records ("skip", "overwrite", "merge")
    • :ids - Only the rows with these key values (1 to 1000)
    • :id_range - {start, end}: only the rows of an integer key range; one bound may be nil, not both
    • :actor_uuid - The user who started the pull; recorded on the transfer (initiated_by_uuid) and its activity row
    • :password - The sender connection's download password, when it has one
    • :on_rate_limited - A function of one argument, called with the seconds before each wait when the sender answers 429 (over the connection's requests per minute). fetch_sender_tables/2, fetch_table_schema/3 and fetch_table_records/3 take it too

A request the sender refuses with 429 is sent again after its Retry-After (clamped, within a total budget: PhoenixKitSync.RateLimit); past the budget the pull fails with :rate_limited.

A table larger than the sender's max_records_per_request comes in pages: the sender's answer says where the next one starts, and the pull follows until the table ends. Each timeout is per page.

A filter goes to the sender with the request; an answer that does not carry "filtered": true (a sender from before filters) is not imported.

Returns

  • {:ok, result} - Map with :imported, :skipped, :errors, plus :unknown_columns when some records carried columns the local table lacks (those records count as errors), and truncated: true when the sender stopped at its max_records_per_request limit and gave no cursor to continue from (a sender from before cursors, a key type without one, or one that did not honour its own cursor): the rows that came are imported, but the transfer is recorded as failed. :unsupported_key_type names the key type when the sender said that is why
  • {:error, :offline} - Sender is offline
  • {:error, :invalid_table_name} - The name is not a plain identifier; nothing is recorded or requested
  • {:error, :invalid_column_name} - The sender's data has a key that is not a plain identifier; no row of the table is written
  • {:error, :invalid_response} - The sender's answer could not be read
  • {:error, :invalid_filter} - An empty, oversized or malformed filter (refused before the request), or one the sender refused
  • {:error, :unsupported_key_type} - The sender's key type takes no filter
  • {:error, :sender_ignores_filters} - The sender ignored the filter; nothing is imported
  • {:error, :table_missing_locally} - The table does not exist on this site; nothing is requested from the sender
  • {:error, :no_primary_key} - The local table has no primary key; it is not pulled, since repeat pulls would duplicate rows
  • {:error, :import_failed} - Handling the response raised or exited; the transfer is marked failed and the details are in the log
  • {:error, reason} - Failed to pull

pull_table_data_with_remap(connection, table_name, uuid_remap, opts \\ [])

@spec pull_table_data_with_remap(
  connection_ref(),
  String.t(),
  uuid_remap(),
  keyword()
) ::
  {:ok, pull_result(), uuid_remap()} | {:error, term(), uuid_remap()}

Same as pull_table_data but accepts and returns a uuid_remap for FK remapping across tables. Returns {:ok, import_result, updated_remap} or {:error, reason, unchanged_remap}, with the same error reasons as pull_table_data/3.

query_sender_status(connection, opts \\ [])

@spec query_sender_status(connection_ref(), keyword()) ::
  {:ok, String.t() | :offline | :not_found} | {:error, term()}

Queries the sender for the current connection status.

Called by receiver to sync their status with the sender's status.

Parameters

  • connection - The receiver connection (must have site_url and auth_token_hash)
  • opts - Options:
    • :timeout - HTTP request timeout (default: 30_000ms)

Returns

  • {:ok, status} - Current status from sender ("active", "suspended", "revoked")
  • {:ok, :offline} - Sender is offline
  • {:ok, :not_found} - Connection not found on sender
  • {:error, reason} - Failed to query

remote_url_prefix()

@spec remote_url_prefix() :: String.t()

Returns the URL prefix the notifier prepends to remote sync API paths.

The remote site mounts its sync routes under PhoenixKit.Config.get_url_prefix/0 (via PhoenixKitSync.Routes.generate/1), so by default we mirror the local site's configured prefix — correct whenever both deployments share the same routing config, which is the common case. (Previously this was hardcoded to /phoenix_kit, which 404'd against any remote mounted under a different prefix — see https://github.com/BeamLabEU/phoenix_kit_sync/issues/8.)

Deployments whose remote uses a different prefix than the local site can override it globally:

config :phoenix_kit_sync, remote_url_prefix: "/custom"

The result is normalized: a leading slash is ensured, trailing slashes are stripped, and an empty or root prefix ("" or "/") yields "" so paths resolve to <site_url>/sync/api/....

verify_connection(connection, opts \\ [])

@spec verify_connection(connection_ref(), keyword()) ::
  {:ok, :exists | :not_found | :offline} | {:error, term()}

Verifies a connection still exists on the remote site.

Called by sender to check if receiver still has the connection. If not, the sender should delete their own connection.

Returns

  • {:ok, :exists} - Connection exists on remote
  • {:ok, :not_found} - Connection was deleted on remote
  • {:ok, :offline} - Remote site is offline
  • {:error, reason} - Failed to verify