@spec apply_remote_status(PhoenixKitSync.Connection.t(), String.t()) ::
  {:ok, PhoenixKitSync.Connection.t()}
  | {:error, :status_change_not_allowed | Ecto.Changeset.t()}

Applies a status the remote site reported for one of this site's connections (the update-status API).

Every move goes through Connection.remote_status_changeset/2, the one place that says what a peer may change: it can mirror the sender's suspension or revocation onto an active "receiver" row and lift a suspension it reported itself, and nothing else. It cannot activate a pending row, undo what an admin here set, or touch a "sender" row.

Returns {:ok, connection} (also when the row already has that status) or {:error, :status_change_not_allowed}.

@spec approve_connection(PhoenixKitSync.Connection.t(), keyword() | String.t() | nil) ::
  {:ok, PhoenixKitSync.Connection.t()}
  | {:error, :status_change_not_allowed | Ecto.Changeset.t()}

Approves a pending connection, making it active.

Options

• :actor_uuid - The user approving the connection

Examples

{:ok, conn} = Connections.approve_connection(conn, actor_uuid: current_user.uuid)

@spec authorize_request(PhoenixKitSync.Connection.t(), String.t() | nil, keyword()) ::
  :ok | {:error, atom() | {:rate_limited, pos_integer()}}

Decides whether a request made under connection may proceed.

The one authorization every entry point shares: the REST actions, the WebSocket upgrade and each message on an open socket call it with the row as it is in the database now, so a restriction holds on all of them alike.

Checks, in order: the row is a "sender" one (a "receiver" row is this site's own pass to the peer's data and gives the peer nothing here), status is "active", not past expires_at, within max_downloads and max_records_total, the client IP is in the whitelist (when one is set), the hour is within the allowed hours, the request is within rate_limit_requests_per_minute (PhoenixKitSync.RateLimit; every call that gets this far counts one), and the download password matches (when one is set).

The two limits are read here from the row as loaded, which refuses a connection that has used them up before any work is done. It does not hold them against requests arriving together: an answer that carries rows takes its download and its records with reserve_transfer/2.

Parameters

• connection - The connection, freshly loaded
• client_ip - The client's address as a string, or nil when unknown (refused when a whitelist is set)
• opts:
  • :password - The download password the client supplied
  • :continuation - true for a further page of a download already counted, which is not held to max_downloads again
  • :rate_limit - false for a socket frame that is not a request (the join, a heartbeat), which is not counted against the minute

Returns

:ok, or {:error, reason} with :unauthorized, :connection_not_active, :connection_expired, :download_limit_reached, :record_limit_reached, :ip_not_allowed, :outside_allowed_hours, {:rate_limited, seconds} (the seconds until the minute frees) or :invalid_password.

@spec cancel_transfer(reservation()) ::
  {:ok, PhoenixKitSync.Connection.t()} | {:error, :not_found}

Gives a reservation back whole: the request it was made for sent nothing.

@spec count_connections(keyword()) :: non_neg_integer()

Counts connections with optional filters.

Options

• :direction - Filter by direction
• :status - Filter by status

@spec create_connection(map()) ::
  {:ok, PhoenixKitSync.Connection.t(), String.t()}
  | {:error, Ecto.Changeset.t()}

Creates a new connection with a generated auth token.

Returns both the connection and the raw auth token (token is only shown once).

Parameters

• attrs - Connection attributes:
  • :name (required) - Human-readable name
  • :direction (required) - "sender" or "receiver"
  • :site_url (required) - URL of the remote site
  • :approval_mode - "auto_approve", "require_approval", "per_table"
  • :created_by - User ID who created the connection
  • Other optional settings

Examples

{:ok, conn, token} = Connections.create_connection(%{
  name: "Staging Server",
  direction: "sender",
  site_url: "https://staging.example.com",
  created_by_uuid: current_user.uuid
})

# Token is only returned once — store it securely.

Deletes a connection.

Examples

{:ok, conn} = Connections.delete_connection(conn)

@spec expiring_soon(non_neg_integer()) :: [PhoenixKitSync.Connection.t()]

Gets connections expiring soon (within given hours).

Useful for sending expiration warnings.

Examples

expiring = Connections.expiring_soon(24)  # Expiring within 24 hours

@spec find_by_hash_and_direction(String.t(), String.t()) ::
  PhoenixKitSync.Connection.t() | nil

Finds a connection by auth token hash and direction.

The auth_token_hash is unique per connection pair, so this can be used to find a specific connection without needing the site URL.

@spec find_by_site_url(String.t(), String.t()) :: PhoenixKitSync.Connection.t() | nil

Finds a connection by site URL and direction.

The URL is compared in its normal form (Connection.normalize_site_url/1), so the spelling a row was stored with and the spelling asked for need not be the same string. When several stored rows are spellings of one site, the one stored exactly as asked wins, then the oldest.

Examples

conn = Connections.find_by_site_url("https://example.com", "sender")

@spec find_by_site_url_and_hash(String.t(), String.t()) ::
  PhoenixKitSync.Connection.t() | nil

Finds a connection by site URL and auth token hash.

Used to identify a specific connection when the remote site requests deletion. The URL is compared as find_by_site_url/2 compares it.

Examples

conn = Connections.find_by_site_url_and_hash("https://example.com", "abc123hash")

@spec find_by_site_url_and_hash(String.t(), String.t(), String.t()) ::
  PhoenixKitSync.Connection.t() | nil

Finds a connection by site URL, auth token hash and direction.

For requests only one side of a connection may make: a status the sender reports names this site's "receiver" row, never a "sender" one. The URL is compared as find_by_site_url/2 compares it.

@spec find_by_token(String.t()) ::
  {:ok, PhoenixKitSync.Connection.t()} | {:error, :invalid_token}

Finds a connection by auth token.

Examples

{:ok, conn} = Connections.find_by_token(token)

@spec get_connection(term()) :: PhoenixKitSync.Connection.t() | nil

Gets a connection by UUID.

Accepts:

• UUID string: get_connection("01234567-89ab-cdef-0123-456789abcdef")

@spec get_connection!(String.t()) :: PhoenixKitSync.Connection.t()

Gets a connection by UUID, raising if not found.

Accepts same inputs as get_connection/1.

@spec list_connections(keyword()) :: [PhoenixKitSync.Connection.t()]

Lists all connections with optional filters.

Options

• :direction - Filter by direction ("sender" or "receiver")
• :status - Filter by status
• :limit - Maximum results
• :offset - Number of results to skip
• :preload - Associations to preload

Examples

connections = Connections.list_connections(direction: "sender", status: "active")

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

PubSub topic that Connections broadcasts on for connection lifecycle events (:connection_created, :connection_updated, :connection_status_changed, :connection_deleted).

Subscribers should reference this constant rather than the literal string so the topic name stays consistent across broadcast and subscribe sites.

@spec reactivate_connection(PhoenixKitSync.Connection.t(), keyword()) ::
  {:ok, PhoenixKitSync.Connection.t()}
  | {:error, :status_change_not_allowed | Ecto.Changeset.t()}

Reactivates a suspended connection.

Examples

{:ok, conn} = Connections.reactivate_connection(conn)

@spec record_transfer(PhoenixKitSync.Connection.t(), map()) ::
  {:ok, PhoenixKitSync.Connection.t()} | {:error, :not_found}

Records a transfer and updates connection statistics.

Adds to the counters without looking at the limits. An answer that carries rows to a peer is counted through reserve_transfer/2 and settle_transfer/3 instead, which is what holds max_downloads and max_records_total.

Parameters

• connection - The connection to update
• attrs - Transfer statistics:
  • :records_count - Number of records transferred
  • :bytes_count - Bytes transferred
  • :downloads - Downloads this counts as (default 1; 0 for a further page of a download already counted)

Examples

{:ok, conn} = Connections.record_transfer(conn, %{
  records_count: 500,
  bytes_count: 250000
})

@spec regenerate_token(PhoenixKitSync.Connection.t(), keyword()) ::
  {:ok, PhoenixKitSync.Connection.t(), String.t()}
  | {:error, Ecto.Changeset.t()}

Regenerates the auth token for a connection.

Returns the new raw token (only shown once).

Examples

{:ok, conn, new_token} = Connections.regenerate_token(conn)

@spec register_incoming(map(), keyword()) ::
  {:ok, PhoenixKitSync.Connection.t()}
  | {:error, :connection_exists | :too_many_pending | Ecto.Changeset.t()}

Creates the receiver connection of an incoming registration, unless this site already has one for that sender or holds :max_pending pending ones.

The checks and the insert run under one advisory lock, so registrations arriving together cannot each pass a check the others are about to make false: two for one site make one row, and the pending ones stop at the cap. The site is compared by its normal form (Connection.normalize_site_url/1).

attrs are create_connection/1's, string-keyed. Returns {:error, :connection_exists}, {:error, :too_many_pending} or the changeset.

@spec remaining_downloads(PhoenixKitSync.Connection.t()) ::
  non_neg_integer() | :unlimited

Gets the remaining download allowance for a connection.

Returns :unlimited if no limit is set.

Examples

Connections.remaining_downloads(conn)
# => 45 (if max_downloads: 50, downloads_used: 5)
# => :unlimited (if max_downloads: nil)

@spec remaining_records(PhoenixKitSync.Connection.t()) ::
  non_neg_integer() | :unlimited

Gets the remaining record allowance for a connection.

Returns :unlimited if no limit is set.

Examples

Connections.remaining_records(conn)
# => 9500 (if max_records_total: 10000, records_downloaded: 500)
# => :unlimited (if max_records_total: nil)

@spec reserve_transfer(PhoenixKitSync.Connection.t(), %{
  downloads: 0 | 1,
  records: pos_integer()
}) ::
  {:ok, reservation()}
  | {:error, :download_limit_reached | :record_limit_reached | :not_found}

Reserves a download and a page of records against the connection's limits.

The limits are held here and nowhere else: one conditional UPDATE adds the download and the page only while max_downloads and max_records_total allow, so two requests arriving together cannot both take the last download or the last records. authorize_request/3 reads the same limits earlier only to refuse sooner.

Every answer that carries rows is reserved first, read with the granted count as its limit, and then settled (settle_transfer/3) with the rows it really sent, or cancelled (cancel_transfer/1) when it sent none.

The guarantee: the rows served under a connection never exceed max_records_total, and the downloads never exceed max_downloads. A request in flight holds its whole page against the record limit until it settles, so records_downloaded can read above the limit for that moment (a request arriving then is refused), and stays charged if the serving process dies before settling.

Parameters

• connection - The connection (only its uuid is read)
• attrs:
  • :downloads - 1 for a new download, 0 for a further page of one already counted
  • :records - The most rows the answer may carry

Returns

• {:ok, reservation} - reservation.records is how many rows may be sent: :records, or what was left of the limit when that is less

• {:error, :download_limit_reached | :record_limit_reached | :not_found}

@spec revoke_connection(PhoenixKitSync.Connection.t(), keyword() | String.t() | nil) ::
  {:ok, PhoenixKitSync.Connection.t()}
  | {:error, :status_change_not_allowed | Ecto.Changeset.t()}

Revokes a connection permanently.

Options

• :actor_uuid - The user revoking the connection
• :reason - Optional reason for revocation

Examples

{:ok, conn} =
  Connections.revoke_connection(conn, actor_uuid: current_user.uuid, reason: "Compromised")

@spec settle_transfer(reservation(), non_neg_integer(), non_neg_integer()) ::
  {:ok, PhoenixKitSync.Connection.t()} | {:error, :not_found}

Settles a reservation with what the answer carried.

Gives back the part of the reserved page that was not sent and adds the transfer to the connection's statistics.

• served - Number of rows sent
• bytes_count - Bytes sent (default 0)

@spec suspend_connection(PhoenixKitSync.Connection.t(), keyword() | String.t() | nil) ::
  {:ok, PhoenixKitSync.Connection.t()}
  | {:error, :status_change_not_allowed | Ecto.Changeset.t()}

Suspends an active connection.

Options

• :actor_uuid - The user suspending the connection
• :reason - Optional reason for suspension

Examples

{:ok, conn} =
  Connections.suspend_connection(conn, actor_uuid: current_user.uuid, reason: "Security audit")

@spec table_allowed?(PhoenixKitSync.Connection.t(), String.t()) :: boolean()

Checks if a table is allowed for this connection.

Takes into account both allowed_tables and excluded_tables.

Examples

Connections.table_allowed?(conn, "users")
# => true

Connections.table_allowed?(conn, "secrets")
# => false

@spec touch_connected(PhoenixKitSync.Connection.t()) ::
  {:ok, PhoenixKitSync.Connection.t()} | {:error, Ecto.Changeset.t()}

Updates last connected timestamp.

Examples

{:ok, conn} = Connections.touch_connected(conn)

@spec update_connection(PhoenixKitSync.Connection.t(), map(), keyword()) ::
  {:ok, PhoenixKitSync.Connection.t()} | {:error, Ecto.Changeset.t()}

Updates a connection's settings.

Examples

{:ok, conn} = Connections.update_connection(conn, %{
  name: "New Name",
  max_downloads: 100
})

@spec validate_connection(String.t(), String.t() | nil, keyword()) ::
  {:ok, PhoenixKitSync.Connection.t()}
  | {:error, atom() | {:rate_limited, pos_integer()}}

Validates an auth token and returns the connection if valid.

Performs comprehensive validation including:

• Token verification
• Status check (must be active)
• Expiration check
• Download limits check
• Record limits check
• IP whitelist check (if provided)
• Time-of-day restrictions check
• Download password check (when the connection has one)

Only a "sender" connection is found: the token of a "receiver" row authorizes nothing on this site. The checks are authorize_request/3's, and opts goes to it.

Parameters

• token - The auth token to validate
• client_ip - The client's IP address (optional)

Returns

• {:ok, connection} - Connection is valid and ready to use
• {:error, reason} - Validation failed with reason:
  • :invalid_token - Token doesn't match any connection
  • :connection_not_active - Connection status is not "active"
  • :connection_expired - Connection has expired
  • :download_limit_reached - Max downloads exceeded
  • :record_limit_reached - Max records exceeded
  • :ip_not_allowed - Client IP not in whitelist
  • :outside_allowed_hours - Current time outside allowed hours
  • :invalid_password - Download password missing or wrong
  • {:rate_limited, seconds} - Over the connection's requests per minute

Examples

case Connections.validate_connection(token, client_ip) do
  {:ok, conn} -> proceed_with_transfer(conn)
  {:error, :connection_expired} -> send_error("Connection has expired")
  {:error, reason} -> send_error("Access denied: #{reason}")
end

@spec validate_download_password(PhoenixKitSync.Connection.t(), String.t() | nil) ::
  :ok | {:error, :invalid_password}

Validates a download password for a connection.

Examples

case Connections.validate_download_password(conn, password) do
  :ok -> proceed()
  {:error, :invalid_password} -> deny_access()
end