Audience Member Sync

The audience member sync system gives packages a standard way to pull external data (CRM membership status, donor records, email lists, etc.) into Allegro and keep it fresh. Providers register once; the platform handles scheduling, queueing, login-path freshness, and failure isolation.

How It Works

Each external integration is a sync provider — a class implementing Allegro\Contracts\SyncProvider. Providers declare:

• What to sync — a membersToSync() query scoping which audience members are in scope for scheduled runs.
• How fresh — a refreshInterval() for scheduled cadence, and an optional maxAgeOnLogin() for how stale data can be before a login triggers a synchronous sync.
• When to run — the platform runs a single global 5-minute heartbeat (allegro-audience:sync:tick) that fans out a per-tenant evaluation job. Per provider, the evaluator checks the audience_sync_runs row for the last dispatch time against refreshInterval() before dispatching a new set of sync jobs.

At JWT generation time, SyncProfilesOrchestrator::ensureFreshForLogin() runs synchronously before the JsonWebTokenGenerating event fires. Listeners that add membership claims to the JWT are therefore guaranteed up-to-date data. Scheduled runs page through the member base with DispatchAudienceSyncChunkJob and enqueue work onto the bulk queue — one rate-limited SyncAudienceMemberBatchJob per 50 members for providers implementing PoolableSyncProvider, or one SyncAudienceMemberJob per member otherwise.

---

Quick Start

Extend Allegro\Services\AudienceSync\AbstractSyncProvider — it provides sensible defaults so you only override what you need.

namespace Acme\Sync;

use Allegro\Models\AudienceMember;
use Allegro\Services\AudienceSync\AbstractSyncProvider;
use Carbon\CarbonInterval;
use Illuminate\Database\Eloquent\Builder;

class AcmeMembershipSync extends AbstractSyncProvider
{
    public function name(): string
    {
        return 'acme_membership'; // matches the audience_member_external_profiles.provider key
    }

    public function refreshInterval(): CarbonInterval
    {
        return CarbonInterval::hours(6);
    }

    public function maxAgeOnLogin(): ?CarbonInterval
    {
        // Sync synchronously on login when the row is older than 30 minutes.
        // Return null to make this a background-only provider.
        return CarbonInterval::minutes(30);
    }

    public function membersToSync(): Builder
    {
        // Only queue jobs for members active in the last 30 days.
        return $this->membersLoggedInWithin(30);
    }

    public function sync(AudienceMember $member): void
    {
        $data = app(AcmeApiClient::class)->fetchMember($member->email);

        $member->externalProfiles()->updateOrCreate(
            ['provider' => $this->name()],
            ['data' => $data],
        );

        // Grant or revoke entitlements based on $data...
    }
}

AbstractSyncProvider defaults:

Method	Default
maxAgeOnLogin()	null (background-only)
isAvailableForTenant()	true

---

Registering a Provider

Call SyncProviderRegistry::register() once in your package's ServiceProvider::boot():

use Allegro\Services\AudienceSync\SyncProviderRegistry;

public function boot(SyncProviderRegistry $registry): void
{
    $registry->register(AcmeMembershipSync::class);
}

register() is idempotent — calling it twice with the same class is safe (e.g., under Octane). Registering two different classes with the same name() throws DuplicateSyncProviderException at boot time.

---

Scoping Members

membersLoggedInWithin(int $days)

AbstractSyncProvider provides this helper for the common case:

public function membersToSync(): Builder
{
    return $this->membersLoggedInWithin(30); // members who logged in within 30 days
}

It queries audience_members.last_active_at, a denormalized column updated whenever a member authenticates or refreshes their JWT. Use this to avoid dispatching jobs for inactive members.

Custom queries

Return any Builder<AudienceMember> from membersToSync(). Return a query that matches zero rows (e.g., AudienceMember::query()->whereRaw('0 = 1')) to disable scheduled sync entirely while still allowing on-login sync.

---

Staleness Configuration

Two knobs control freshness:

Method	Type	Purpose
refreshInterval()	CarbonInterval	How old a row can be before the scheduled job re-syncs it
maxAgeOnLogin()	?CarbonInterval	How old a row can be before a login triggers a synchronous sync; null = never sync on login

Background-only provider

Return null from maxAgeOnLogin() (the default). The provider runs on the scheduler cadence only. Use when upstream latency is too high for the login path, or when data does not affect the JWT.

public function maxAgeOnLogin(): ?CarbonInterval
{
    return null; // background only; login never blocks on this provider
}

Combined (most common)

Use both: schedule a background refresh and also sync on login when data is stale relative to maxAgeOnLogin(). The login threshold is typically tighter than refreshInterval() so a member always gets fresh data even if the last scheduled run was hours ago.

---

Tenant Availability

Override isAvailableForTenant() to gate a provider on tenant-level configuration:

public function isAvailableForTenant(): bool
{
    return config('acme.enabled', false);
}

The platform evaluates this predicate at every entry point: the login orchestrator, the Artisan command, and inside the queued job. A provider that returns false here is silently skipped — no error, no job dispatched.

---

Side Effects in sync()

sync() may do more than write the primary audience_member_external_profiles row. Common patterns:

Write member meta:

$member->meta()->updateOrCreate(
    ['key' => 'total_donations'],
    ['value' => $data['total_donations']],
);

Grant or revoke entitlements:

if ($data['is_active_member']) {
    $member->entitlements()->updateOrCreate(
        ['source_provider' => $this->name(), 'resource_id' => $resourceId],
        ['expires_at' => null],
    );
} else {
    $member->entitlements()
        ->where('source_provider', $this->name())
        ->delete();
}

Idempotency is required

sync() may be called multiple times for the same member. Use upserts keyed on source_provider for entitlement writes and updateOrCreate for profile rows. Wrap writes in a DB::transaction() so a throw mid-sync rolls back partial state.

---

Login Path

JsonWebToken::generate() calls SyncProfilesOrchestrator::ensureFreshForLogin($member) before dispatching JsonWebTokenGenerating. The orchestrator:

1. Iterates providers registered for the current tenant where maxAgeOnLogin() !== null.
2. For each, checks whether the primary audience_member_external_profiles row is younger than maxAgeOnLogin().
3. If stale or missing, calls sync() synchronously — taking the same Cache::lock used by SyncAudienceMemberJob to prevent concurrent writes.
4. Catches and logs any exceptions (provider failures as warning, unexpected infrastructure failures as error). The JWT is issued regardless of sync failures.

The JsonWebTokenGenerating event fires only after all login-path providers have run, so JWT claim listeners always see the freshest available data.

---

Scheduled Runs

Scheduling is driven by a single global entry — there is no per-provider cron registration. The platform registers one scheduler entry in routes/console.php:

allegro-audience:sync:tick   →   every 5 minutes

Provider authors implement refreshInterval() as the single source of truth for cadence. Do not implement a scheduleCron() method — it was removed from the contract. The scheduler does not look at your provider individually.

How the tick works

1. allegro-audience:sync:tick iterates every registered tenant and dispatches an EvaluateTenantSyncsJob for each via $tenant->execute(...), so the evaluation runs bound to that tenant's context.
2. Inside the job, for each provider returned by SyncProviderRegistry::availableForTenant():
   • The job loads (or creates) the audience_sync_runs row keyed on provider. The table is tenant-scoped via UsesTenantConnection, so each tenant has its own row per provider.
   • Mid-flight guard — a non-null last_synced_id means a paging chain is running; a fresh heartbeat (updated_at) leaves it alone, a stale one resumes the chain from the saved cursor.
   • Due-check guard — if last_dispatched_at + refreshInterval() is still in the future, the provider isn't due yet; skip.
   • Otherwise the job claims the run window (last_dispatched_at = now()) and starts a DispatchAudienceSyncChunkJob chain. The chain pages through membersToSync() with a keyset cursor, 10,000 members per page, and enqueues the page's sync jobs via Bus::bulk — batch jobs for poolable providers, per-member jobs otherwise — before chaining itself to the next page.
3. Workers on the bulk queue then process the sync jobs. Each job re-checks freshness per member before syncing, so a synchronous login sync between dispatch and execution is respected — no double-sync.

The evaluator does not track run completion. Gating on last_dispatched_at + refreshInterval() alone means the worst case is a re-dispatch before the prior run has fully drained — harmless because syncs are idempotent per (member, provider) and skipped when the profile is fresh. The trade-off avoids the stuck-forever failure mode where a missing "finished" signal would permanently block re-evaluation.

Continuous syncing

Everything above describes the default burst pacing: a provider claims its run window and the chunk chain pages the member base as fast as backpressure and the rate limiter allow, finishing well inside the interval and idling out the rest of it.

Continuous mode changes the rate and nothing else. It is opt-in per tenant — the switch is on the landlord tenant's Audience Sync page, and it writes the audience_sync.continuous tenant setting:

TenantSetting::set(ContinuousSyncPlanner::SETTING_KEY, true);

With it on, EvaluateTenantSyncsJob claims the same run window against the same audience_sync_runs row, but instead of starting a self-chaining DispatchAudienceSyncChunkJob, each tick advances the cursor itself by ContinuousSyncPlanner::sliceSize() members:

membersToSync()->where('id', '>', $run->last_synced_id)->orderBy('id')->limit($slice)

That is the burst pager's own query, so a lap covers exactly the members the provider asked for. The slice is the provider's member count divided by the number of 5-minute ticks in its refreshInterval(), plus half again as slack so a lap finishes a little early rather than drifting past the interval it is meant to fit in. The count is cached for an hour — it only paces the lap — except for a zero, which is never cached, so a provider whose window fills up after an empty count is not left crawling a member a tick until the entry expires.

An empty page ends the lap and clears the cursor, which hands the provider back to the ordinary interval gate: the next lap opens once refreshInterval() has passed since this one began. So a lap takes about one refresh interval, and laps do not overlap.

Providers need no changes, and everything the sync dashboards read keeps its meaning: the cursor is the lap's position, last_dispatched_at is when the lap started, and the statuses are the real ones.

Why the cursor, and not staleness ordering

Selecting the least-recently-synced members each tick is the obvious design and it has a trap. Ordering by a profile's timestamp means reading from audience_member_external_profiles and then filtering by membersToSync(), and a member who has a profile row but has fallen out of the provider's scope — past the activity window, holding no entitlement — is never selected, so nothing ever advances their timestamp. Those rows accumulate at the head of the ordering permanently. Once there are more of them than one slice's worth, every tick selects nothing but ineligible rows and the sync stops dead, silently, with the run row still looking healthy.

Walking the member query by cursor has no such state to get stuck on. The cursor advances past whatever it selected, ineligible members are never in the result to begin with, and a member whose sync fails is simply retried on the next lap.

Ordering by staleness buys little in exchange: a lap reaches every member once per interval either way, and a member synced on login meanwhile costs one row in a SELECT rather than an API call, because the sync jobs re-check freshness per member before calling out.

Concurrency and backpressure

Ticks for one tenant can overlap when the bulk queue is congested, and two of them would read the same cursor and dispatch the same members twice. The run row is read, advanced and saved inside a per-tenant, per-provider cache lock — the read included, so an overlapping tick cannot act on a cursor that has already moved; a tick that cannot take the lock waits for the next one.

The slice reaches the queue before the cursor moves past it, so an enqueue that fails is retried from the same cursor on the next tick rather than leaving those members unsynced for the rest of the lap. A tick is skipped outright while the bulk queue holds more than MAX_QUEUE_BACKLOG pending jobs, the same backpressure the chunk chain applies: a slice sized off a large member base, or paced against a short refreshInterval(), can still outrun the workers and the rate limiter. The pacing slack absorbs the skipped ticks.

Switching modes

The switch takes effect on the next tick and nothing needs restarting. The paced tick is the cursor's only writer while continuous mode is on: a DispatchAudienceSyncChunkJob that runs — from a chain already in flight when the switch was flipped, or from a manual CLI run — stops without touching the cursor, and the paced ticks take over the rest of the lap from where it stands.

Turning it off hands the cursor back to the chunk chain, which resumes paging it at full speed — but not immediately. A paced lap in progress leaves a fresh heartbeat on the run row, which the evaluator cannot tell from a live chain, so the chain resumes only once that heartbeat crosses the stall window (STALL_AFTER_MINUTES, an hour). Turning the switch off mid-lap therefore costs up to an hour before full-speed paging picks up from the cursor.

Backpressure and failure semantics

The chunk chain is built so a very large member base (1M+) or a failing page can never flood the queue:

• Backpressure — before paging, the chunk job checks the bulk queue depth. Above MAX_QUEUE_BACKLOG pending jobs it defers itself for a few minutes (bumping the run heartbeat so the evaluator doesn't mistake the pause for a stall) instead of enqueueing another page. A run therefore never sits more than a couple of pages ahead of what the workers and rate limiter drain, regardless of member count.
• Transient failures — the chunk job retries in place (3 tries, 30s backoff). Re-running a page is safe: dispatch is idempotent downstream.
• Terminal failures — after exhausting retries, failed() clears the cursor and reports a descriptive exception. The run ends visibly instead of the evaluator resuming the same failing page every stall window, and the already-claimed last_dispatched_at defers the next full run to the provider's regular interval.
• Silent deaths (deploy kill, OOM) never reach failed(); those are the case the evaluator's stall-gate resume exists for.

Because cadence is just refreshInterval() + the row-based guards, a provider that wants "daily" simply returns CarbonInterval::day(); one that wants "every 15 minutes" returns CarbonInterval::minutes(15). The next tick (within 5 minutes) will pick it up.

Running manually

The manual CLI still works for backfills and debugging. It bypasses the refresh-interval throttle, claims the same audience_sync_runs row, and starts the same DispatchAudienceSyncChunkJob chain. Claiming the row is what stops the scheduler double-dispatching the provider for that tenant, and the command refuses to start while a chain is already mid-flight. Under continuous syncing the claim is the whole of a manual run: it opens a lap now, past the interval gate, and the paced ticks walk it — no chain is started, because the cursor must have a single writer.

# Sync all available providers for one tenant
php artisan allegro-audience:sync --tenant=42

# Sync one provider only for one tenant
php artisan allegro-audience:sync acme_membership --tenant=42

---

Queue Configuration

EvaluateTenantSyncsJob and every member sync job — SyncAudienceMemberJob, SyncAudienceMemberBatchJob and SyncUserAttributeFromExternalProfilesJob — run on the bulk queue, which has its own Laravel Cloud managed queue and its own worker pool — see Queue Management. Locally the queue is covered by the workers composer run dev starts.

DispatchAudienceSyncChunkJob is the exception: it runs on default, above the queue it feeds, so the dispatcher never starves behind the member jobs it enqueues.

Keeping the member jobs off high and default means a CRM outage that backs up retries, or a tick that fans out a job per member across every tenant, cannot delay a login email. bulk runs on the Pro compute class because it carries jobs well past the 90-second Flex cap.

SyncAudienceMemberJob properties (non-poolable providers, one job per member):

Property	Value	Description
$tries	3	Retried up to 3 times before landing in failed_jobs
$backoff	60s	Seconds between retries
$uniqueFor	600s	ShouldBeUnique window — deduplicates queued jobs per (member, provider)

SyncAudienceMemberBatchJob properties (poolable providers, up to 50 members per job, requests pooled 10 at a time):

Property	Value	Description
middleware	RateLimitedWithRedis('audience-sync-batches')	Per-provider batches-per-minute cap (allegro.audience_sync.batches_per_minute, default 10)
retryUntil()	24h	Time-bounded retries — rate-limiter releases consume attempts, so no tries count
$backoff	60s	Seconds between retries

Batch jobs are deliberately not unique: per-job unique lock writes at member volume have exhausted Redis memory before, and duplicates are absorbed by the per-member freshness check.

---

Testing

Use Allegro\Testing\FakeSyncProvider in feature tests:

use Allegro\Services\AudienceSync\SyncProviderRegistry;
use Allegro\Testing\FakeSyncProvider;
use Carbon\CarbonInterval;

beforeEach(function () {
    // Reset the singleton so registrations from real boot don't bleed in.
    app()->forgetInstance(SyncProviderRegistry::class);
});

it('dispatches a sync job for active members', function () {
    $provider = new FakeSyncProvider(
        providerName: 'fake',
        refresh: CarbonInterval::hour(),
        loginAge: CarbonInterval::minutes(30),
    );

    app()->instance(FakeSyncProvider::class, $provider);
    app(SyncProviderRegistry::class)->register(FakeSyncProvider::class);

    // ...assert jobs dispatched, sync call count, etc.
});

SyncProviderRegistry::register() takes a class string and resolves the provider from the container. Bind your configured instance with app()->instance(...) first so the registry resolves your instance instead of constructing a fresh one with defaults.

FakeSyncProvider constructor parameters (in order):

Parameter	Default	Description
$providerName	'fake'	Return value of name()
$refresh	null	refreshInterval(); when null, falls through to CarbonInterval::hour() via refreshInterval()
$loginAge	null	maxAgeOnLogin()
$availableForTenant	true	isAvailableForTenant()
$membersQuery	null	Closure returning a Builder; defaults to empty

$onSync is a public property (not a constructor argument) — assign directly: $fake->onSync = fn ($member) => .... FakeSyncProvider::$syncCallCount tracks how many times sync() was called.

Simulating staleness

Use Carbon::setTestNow() to move time forward past an interval:

use Carbon\Carbon;

it('re-syncs stale data on login', function () {
    $member = AudienceMember::factory()->create();
    $member->externalProfiles()->create([
        'provider' => 'fake',
        'data' => [],
        'updated_at' => now()->subHour(),
    ]);

    Carbon::setTestNow(now()->addHours(2));

    // trigger login path...
});

Asserting dispatched jobs

use Allegro\Jobs\SyncAudienceMemberJob;
use Illuminate\Support\Facades\Queue;

it('dispatches jobs on scheduled run', function () {
    Queue::fake();

    $this->artisan('allegro-audience:sync', ['provider' => 'fake']);

    Queue::assertPushed(SyncAudienceMemberJob::class);
});

---

End-to-End Flow

Login path
----------
JsonWebToken::generate($member)
  └─ SyncProfilesOrchestrator::ensureFreshForLogin($member)
       └─ foreach provider where maxAgeOnLogin() !== null
            └─ EnsureFreshProfile::for($member, $providerName)
                 ├─ row fresh? → return true (no sync)
                 └─ stale/missing → Cache::lock → provider->sync($member)
  └─ event(JsonWebTokenGenerating)  ← listeners see fresh data

Scheduled path
--------------
scheduler fires allegro-audience:sync:tick  (every 5 minutes)
  └─ foreach tenant
       └─ EvaluateTenantSyncsJob (dispatched via $tenant->execute(...))
            └─ foreach provider in registry->availableForTenant()
                 ├─ load audience_sync_runs row
                 ├─ due-check (last_dispatched_at + refreshInterval() <= now)? skip
                 └─ SyncAudienceMemberJob::dispatch per member
                      ├─ update last_dispatched_at = now()
                      └─ worker picks up each job
                           └─ Cache::lock → freshness re-check → provider->sync($member)

The Cache::lock is shared between the login path and the queued job path. If a login sync is in progress when a worker picks up a job (or vice versa), one path blocks for up to 5 seconds then yields.

---

Operations

Force-resyncing a member

From the audience member detail page in the admin UI, the Sync Status panel lists every registered provider with its last-synced timestamp and a Re-sync now button. Clicking it dispatches SyncAudienceMemberJob with the $force = true flag, bypassing the freshness check.

Verifying freshness

Query audience_member_external_profiles for the member and provider:

SELECT provider, updated_at
FROM audience_member_external_profiles
WHERE audience_member_id = ?
ORDER BY provider;

The updated_at column is the staleness anchor. Compare it against the provider's refreshInterval() to determine whether a job will re-sync on the next scheduled run.

Checking sync status for a tenant

Each tenant has a read-only Audience Sync page that reports what the next scheduled tick will do, without you querying the database. Reach it from Global Settings → Tenants → [tenant] → Audience Sync.

The page lists every registered provider with its last run, next run, cadence, the number of members due on the next run (the bulk backlog), and profile freshness. Providers gated behind a package the tenant has not enabled are listed separately, since they never run for that tenant.

Each provider carries an overall status derived from the same audience_sync_runs state the scheduler reads, so it always matches what the next tick will do:

Status	Meaning
Running	A paging run is mid-flight and its heartbeat is recent — the provider is actively syncing members in chunks.
Stalled	A paging run has a saved cursor but no chunk activity for over an hour. The next tick resumes it from the cursor.
Never run	The provider has never been dispatched for this tenant.
Overdue	The refresh window has opened but no dispatch followed — either there is nothing to sync, or the scheduler is not ticking.
Idle	The provider has run, is within its refresh window, and has nothing due yet. This is the healthy resting state.
Inactive	The provider is not available for this tenant (resolved but switched off — for example, no email provider is configured).

note

Profile freshness and the members-due backlog are the expensive columns on a large tenant, so they load a moment after the rest of the page.

Rolling out a new provider

1. Implement SyncProvider (or extend AbstractSyncProvider) in your package.
2. Register in your ServiceProvider::boot().
3. Gate the provider with isAvailableForTenant() if rollout should be gradual.
4. Deploy. The next tick (within 5 minutes) will pick up the provider and evaluate it per tenant.
5. To backfill immediately: php artisan allegro-audience:sync your_provider_name --tenant={id}.

Recovering from failed jobs

Failed jobs land in failed_jobs after 3 retries. Once the upstream service recovers, flush and re-dispatch:

php artisan queue:retry all

Or let the next scheduled run naturally re-queue jobs for members still in scope — failed_jobs can safely be flushed after upstream recovery.