PhoenixKitSync.Connection (PhoenixKitSync v0.2.5)

Schema for DB Sync persistent connections between PhoenixKit instances.

Connections allow two PhoenixKit sites to establish a permanent relationship for data synchronization, replacing ephemeral session codes with token-based authentication.

Direction

Each connection has a direction:

"sender" - This site will send data (configured on the data-sharing site)
"receiver" - This site will receive data (configured on the data-receiving site)

Approval Modes (Sender-side)

Stored, and answered for by requires_approval?/2, but not applied to pulls: no transfer is held for approval, so every mode behaves as "auto_approve".

"auto_approve" - All transfers are automatically approved
"require_approval" - Each transfer needs manual approval
"per_table" - Tables in auto_approve_tables don't need approval, others do

Status Flow

pending → active → suspended → revoked
                ↘
                expired (set by `Connections.expire_connections/0`)

A connection past expires_at or its limits is refused whatever its stored status (Connections.authorize_request/3); nothing in this module schedules expire_connections/0.

Security Features

Checked on every request by Connections.authorize_request/3:

Token-based authentication (auth_token_hash)
Optional download password
IP whitelist
Time-of-day restrictions
Download and record limits
Expiration date

rate_limit_requests_per_minute is stored and not applied.

Usage Examples

# Create a sender connection
{:ok, conn} = Connections.create_connection(%{
  name: "Production Receiver",
  direction: "sender",
  site_url: "https://receiver.example.com",
  approval_mode: "auto_approve"
})

# Approve a pending connection
{:ok, conn} = Connections.approve_connection(conn, admin_user_uuid)

# Suspend a connection
{:ok, conn} = Connections.suspend_connection(conn, admin_user_uuid, "Security audit")

Summary

Types

t()

A connection row. auth_token and download_password are virtual: set on the way in, stored only as their hashes. A *_user association is Ecto.Association.NotLoaded until preloaded.

Functions

active?(conn)

Checks if a connection is currently active and within limits.

approve_changeset(connection, admin_user_uuid)

Changeset for approving a connection.

changeset(connection, attrs)

Creates a changeset for connection creation.

expired?(connection)

Checks if a connection has expired.

generate_auth_token()

Generates a secure random token for connection authentication.

ip_allowed?(connection)

Checks if an IP address is in the whitelist.

ip_allowed?(arg1, ip)

max_records_per_request_range()

The bounds of max_records_per_request, the most rows one request is answered with: the changesets validate against them and the connection form offers them.

normalize_site_url(url)

The form two spellings of one site's URL share, for comparing them: scheme and host in lower case (the host without a trailing dot), the scheme's default port left out, trailing slashes dropped. The path keeps its case, and a fragment is ignored.

reactivate_changeset(connection)

Changeset for reactivating a suspended connection.

remote_status_changeset(connection, status)

Changeset for a status the remote site reports through the API.

remote_suspended?(connection)

Whether the connection's suspension was reported by the remote site rather than set by an admin here.

requires_approval?(arg1, table)

Checks if a table requires approval based on the connection's approval mode.

revoke_changeset(connection, admin_user_uuid, reason \\ nil)

Changeset for revoking a connection.

settings_changeset(connection, attrs)

Changeset for updating connection settings.

stats_changeset(connection, attrs)

Changeset for updating connection statistics.

suspend_changeset(connection, admin_user_uuid, reason \\ nil)

Changeset for suspending a connection.

table_allowed?(connection, table)

Checks if a table is allowed to be accessed.

verify_auth_token(arg1, token)

Verifies an auth token against the stored hash.

verify_download_password(arg1, password)

Verifies a download password against the stored hash.

within_allowed_hours?(connection)

Checks if current time is within allowed hours.

within_download_limits?(connection)

Checks if a connection is within download limits.

within_record_limits?(connection)

Checks if a connection is within record limits.

Types

t()

@type t() :: %PhoenixKitSync.Connection{
  __meta__: term(),
  allowed_hours_end: integer() | nil,
  allowed_hours_start: integer() | nil,
  allowed_tables: [String.t()] | nil,
  approval_mode: String.t() | nil,
  approved_at: DateTime.t() | nil,
  approved_by_user:
    PhoenixKit.Users.Auth.User.t() | Ecto.Association.NotLoaded.t() | nil,
  approved_by_uuid: UUIDv7.t() | nil,
  auth_token: String.t() | nil,
  auth_token_hash: String.t() | nil,
  auto_approve_tables: [String.t()] | nil,
  auto_sync_enabled: boolean() | nil,
  auto_sync_interval_minutes: integer() | nil,
  auto_sync_tables: [String.t()] | nil,
  created_by_user:
    PhoenixKit.Users.Auth.User.t() | Ecto.Association.NotLoaded.t() | nil,
  created_by_uuid: UUIDv7.t() | nil,
  default_conflict_strategy: String.t() | nil,
  direction: String.t() | nil,
  download_password: String.t() | nil,
  download_password_hash: String.t() | nil,
  downloads_used: integer() | nil,
  excluded_tables: [String.t()] | nil,
  expires_at: DateTime.t() | nil,
  inserted_at: DateTime.t() | nil,
  ip_whitelist: [String.t()] | nil,
  last_connected_at: DateTime.t() | nil,
  last_transfer_at: DateTime.t() | nil,
  max_downloads: integer() | nil,
  max_records_per_request: integer() | nil,
  max_records_total: integer() | nil,
  metadata: map() | nil,
  name: String.t() | nil,
  rate_limit_requests_per_minute: integer() | nil,
  records_downloaded: integer() | nil,
  revoked_at: DateTime.t() | nil,
  revoked_by_user:
    PhoenixKit.Users.Auth.User.t() | Ecto.Association.NotLoaded.t() | nil,
  revoked_by_uuid: UUIDv7.t() | nil,
  revoked_reason: String.t() | nil,
  site_url: String.t() | nil,
  status: String.t() | nil,
  suspended_at: DateTime.t() | nil,
  suspended_by_user:
    PhoenixKit.Users.Auth.User.t() | Ecto.Association.NotLoaded.t() | nil,
  suspended_by_uuid: UUIDv7.t() | nil,
  suspended_reason: String.t() | nil,
  total_bytes_transferred: integer() | nil,
  total_records_transferred: integer() | nil,
  total_transfers: integer() | nil,
  updated_at: DateTime.t() | nil,
  uuid: UUIDv7.t() | nil
}

A connection row. auth_token and download_password are virtual: set on the way in, stored only as their hashes. A *_user association is Ecto.Association.NotLoaded until preloaded.

Functions

active?(conn)

@spec active?(term()) :: boolean()

Checks if a connection is currently active and within limits.

approve_changeset(connection, admin_user_uuid)

@spec approve_changeset(t(), String.t() | map() | nil) :: Ecto.Changeset.t()

Changeset for approving a connection.

changeset(connection, attrs)

@spec changeset(t() | Ecto.Changeset.t(), map()) :: Ecto.Changeset.t()

Creates a changeset for connection creation.

expired?(connection)

@spec expired?(t()) :: boolean()

Checks if a connection has expired.

generate_auth_token()

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

Generates a secure random token for connection authentication.

ip_allowed?(connection)

@spec ip_allowed?(t()) :: boolean()

Checks if an IP address is in the whitelist.

ip_allowed?(arg1, ip)

@spec ip_allowed?(term(), term()) :: boolean()

max_records_per_request_range()

@spec max_records_per_request_range() :: Range.t()

The bounds of max_records_per_request, the most rows one request is answered with: the changesets validate against them and the connection form offers them.

normalize_site_url(url)

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

Connections store site_url as it was given; this is what registration and the Connections.find_by_site_url* lookups compare, so https://Example.com:443/ and https://example.com are one site. Text that is not a URL with a scheme and a host is returned trimmed.

reactivate_changeset(connection)

@spec reactivate_changeset(t()) :: Ecto.Changeset.t()

Changeset for reactivating a suspended connection.

remote_status_changeset(connection, status)

@spec remote_status_changeset(t(), String.t()) ::
  {:ok, Ecto.Changeset.t()} | :unchanged | {:error, :status_change_not_allowed}

Changeset for a status the remote site reports through the API.

The peer is not trusted with this site's decisions, so only these moves exist, and only on a "receiver" row (the sender is the side that notifies):

active to suspended or revoked
suspended to active or revoked, when the suspension itself came from the peer (a suspension set here stays until an admin here lifts it)

A pending, revoked or expired row never changes, and neither does a "sender" row: those statuses decide what the peer may read.

Returns {:ok, changeset}, :unchanged when the row already has that status, or {:error, :status_change_not_allowed}.

remote_suspended?(connection)

@spec remote_suspended?(t()) :: boolean()

Whether the connection's suspension was reported by the remote site rather than set by an admin here.

requires_approval?(arg1, table)

@spec requires_approval?(term(), term()) :: boolean()

Checks if a table requires approval based on the connection's approval mode.

revoke_changeset(connection, admin_user_uuid, reason \\ nil)

@spec revoke_changeset(t(), String.t() | map() | nil, String.t() | nil) ::
  Ecto.Changeset.t()

Changeset for revoking a connection.

settings_changeset(connection, attrs)

@spec settings_changeset(t(), map()) :: Ecto.Changeset.t()

Changeset for updating connection settings.

stats_changeset(connection, attrs)

@spec stats_changeset(t(), map()) :: Ecto.Changeset.t()

Changeset for updating connection statistics.

suspend_changeset(connection, admin_user_uuid, reason \\ nil)

@spec suspend_changeset(t(), String.t() | map() | nil, String.t() | nil) ::
  Ecto.Changeset.t()

Changeset for suspending a connection.

table_allowed?(connection, table)

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

Checks if a table is allowed to be accessed.

verify_auth_token(arg1, token)

@spec verify_auth_token(term(), term()) :: boolean()

Verifies an auth token against the stored hash.

verify_download_password(arg1, password)

@spec verify_download_password(term(), term()) :: boolean()

Verifies a download password against the stored hash.

within_allowed_hours?(connection)

@spec within_allowed_hours?(t()) :: boolean()

Checks if current time is within allowed hours.

within_download_limits?(connection)

@spec within_download_limits?(t()) :: boolean()

Checks if a connection is within download limits.

within_record_limits?(connection)

@spec within_record_limits?(t()) :: boolean()

Checks if a connection is within record limits.