docs: describe and diagram the Leantime sync

CHANGELOG.md

@@ -8,6 +8,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+* [PR-333](https://github.com/itk-dev/economics/pull/333)
+  * Documented how Economics synchronizes from the Leantime data-api plugin in `docs/leantime-sync.md`: the
+    scheduled jobs, the command options, and the paging, incrementality and deletion behaviour that is not
+    obvious from the code.
+  * Three flowcharts carry the structure — fetching a page, turning a page into rows, and the plugin side — and
+    a sequence diagram carries one update run end to end. Each owns ground the others do not, so a change lands
+    in one diagram rather than several.
+  * Rewrote the `Synchronization` section in `README.md`, which described a `QueueSyncCommand` and an
+    `app:queue-sync` that do not exist, credited the Symfony Scheduler for work cron does, and named
+    `DataProviderServiceInterface` instead of `DataProviderInterface`.
 * [PR-335](https://github.com/itk-dev/economics/pull/335)
   * Stopped `projectRemovedFromDataProvider()` hard-deleting a project that a version, a project billing or a
     service agreement still points at. Each of those points back with a non-nullable, non-cascading foreign key,

README.md

@@ -52,38 +52,50 @@ accounts.
 
 ## Synchronization
 
-Economics depends on data fra external systems. The integrations with external systems are called Data Providers.
+Economics depends on data from external systems. The integrations with external systems are called Data Providers.
 
-Each Data Provider integration should implement `App\Interface\DataProviderServiceInterface`.
+Each Data Provider integration should implement `App\Interface\DataProviderInterface`. The only
+implementation today is `App\Service\LeantimeApiService`, which pulls from the
+[data-api plugin](https://github.com/itk-dev/data-api) running in Leantime.
 
-The data synchronization is handled by symfony messenger. This is handled differently in development and production.
+Synchronization is a paged pull driven by Symfony Messenger: a command dispatches one message per
+entity type, each message fetches a page of at most 100 rows and re-dispatches itself for the next
+page, and every fetched row becomes an upsert message of its own.
 
-### Production
+See [docs/leantime-sync.md](docs/leantime-sync.md) for the full picture, the scheduled jobs and the
+command options.
 
-Supervisor is added to `docker-compose.server.override.yaml` to make sure the job queue is running.
+### Production
 
-Symfony scheduler is used for creating a new job each hour at minute 5. See `App\Command\QueueSyncCommand`.
+Cron jobs on the server run the sync commands — see the `cron` section of
+`.woodpecker/prod_itk_economics.yml`. Supervisor is added in `docker-compose.server.override.yml` to
+keep a worker consuming the `async` transport.
 
 ### Develop
 
 In development the job queue should be run manually.
 
-```sh
-docker compose exec phpfpm bin/console messenger:consume async -vv --failure-limit 1
+```shell
+task messenger
 ```
 
 ### Queuing jobs
 
-Jobs can be queued manually with App\Command\QueueSyncCommand
+Jobs can be queued manually.
 
-```sh
-docker compose exec phpfpm bin/console app:queue-sync
+```shell
+# Everything modified within the last hour.
+task phpfpm -- bin/console app:data-providers:sync-modified
+
+# Entities deleted within the last hour.
+task phpfpm -- bin/console app:data-providers:sync-deleted
+
+# A full sync of a single entity type, as async jobs, ignoring modified timestamps.
+task phpfpm -- bin/console app:data-providers:sync -j -p -d
 ```
 
 Jobs can also be queued in the admin interface in the bottom left corner.
 
-In production jobs are queued automatically each hour.
-
 ## Development
 
 Getting started:

docs/leantime-sync.md

@@ -0,0 +1,192 @@
+# Synchronization from Leantime
+
+Economics does not talk to Leantime directly. Leantime runs the
+[data-api plugin](https://github.com/itk-dev/data-api), which exposes read-only endpoints under
+`/APIData/API/`, and Economics pulls from those endpoints on a schedule. Nothing is pushed from
+Leantime; every sync starts as a cron job on the Economics host.
+
+The pull is paged, incremental and queue driven: a command dispatches one message per entity type,
+each message fetches one page of at most 100 rows, dispatches one upsert message per row, and
+re-dispatches itself for the next page until a short page ends the run.
+
+The three flowcharts below split that structure into the parts worth looking at separately, and the
+sequence diagram after them follows a single update run end to end. Throughout, a solid arrow is a
+message dispatched or a request sent, and a dashed arrow is a response.
+
+## Fetching a page
+
+```mermaid
+flowchart TB
+    CRON["Cron on the Economics host<br>*/15 · */25 · 02:00–02:40"]
+    CMD["Sync commands<br>app:data-providers:sync…"]
+    SVC["LeantimeApiService<br>one message per entity type,<br>per enabled DataProvider"]
+    Q(["Messenger async transport<br>messenger:consume async"])
+    H["LeantimeUpdateHandler · LeantimeDeleteHandler<br>updateAsJob() · deleteAsJob()"]
+    API["data-api plugin in Leantime"]
+
+    CRON --> CMD --> SVC
+    SVC -- "LeantimeUpdateMessage / LeantimeDeleteMessage<br>start 0, limit 100" --> Q
+    Q --> H
+    H -- "POST /APIData/API/{type}<br>x-api-key, start, limit,<br>modifiedAfter or deletedAfter" --> API
+    API -. "results, resultsCount" .-> H
+    H -- "next page while resultsCount == limit" --> Q
+```
+
+Everything above assumes async handling (`-j`). Without it the same messages are stamped for the
+`sync://` transport and run inline in the cron process instead of going through the queue — which is
+what `app:data-providers:sync-deleted` does, deliberately. Read **Delete ordering** below before
+changing that.
+
+## Turning a page into rows
+
+```mermaid
+flowchart TB
+    H["LeantimeUpdateHandler · LeantimeDeleteHandler"]
+    Q(["Messenger async transport"])
+    UPS["Upsert*Handler<br>project · version · issue · worklog · worker"]
+    RMV["EntityRemovedFromDataProviderHandler"]
+    DPS["DataProviderService<br>creates or updates the entity,<br>hard-deletes it, or marks sourceDeletedDate"]
+    ECDB[("Economics DB<br>Project · Version · Issue · Worklog · Worker")]
+
+    H -- "Upsert*Message, one per row" --> Q
+    H -- "EntityRemovedFromDataProviderMessage,<br>one per deletion" --> Q
+    Q --> UPS --> DPS
+    Q --> RMV --> DPS
+    DPS --> ECDB
+```
+
+## The plugin side
+
+```mermaid
+flowchart TB
+    H["LeantimeUpdateHandler · LeantimeDeleteHandler"]
+    API["Controllers/API.php<br>validates, 400 on a bad parameter"]
+    APIS["Services/APIData"]
+    REPO["Repositories/ApiDataRepository"]
+    LTDB[("Leantime DB<br>zp_projects · zp_tickets · zp_timesheets · zp_user<br>itk_data_api_modified and the itk_*_deleted tables<br>kept current by triggers")]
+
+    H -- "POST /APIData/API/{type}" --> API
+    API --> APIS --> REPO --> LTDB
+    API -. "results, resultsCount" .-> H
+```
+
+## One update run, page by page
+
+```mermaid
+sequenceDiagram
+    autonumber
+    participant Cron
+    participant Service as LeantimeApiService
+    participant Queue as async transport
+    participant Worker as LeantimeUpdateHandler
+    participant API as data-api plugin
+    participant DPS as DataProviderService
+
+    Cron->>Service: updateAll(async, modifiedAfter)
+    loop per entity type, per enabled DataProvider
+        Service->>Queue: LeantimeUpdateMessage(start=0, limit=100)
+    end
+    Queue->>Worker: LeantimeUpdateMessage
+    Worker->>API: POST /APIData/API/tickets<br>{start, limit, modifiedAfter, projectIds}
+    API-->>Worker: {parameters, resultsCount, results}
+    loop per row
+        Worker->>Queue: UpsertIssueMessage(DataProviderIssueData)
+    end
+    alt resultsCount == limit
+        Worker->>Queue: LeantimeUpdateMessage(start = highest id + 1)
+    else resultsCount == limit, no usable id
+        Note over Worker: run stops, error logged
+    else short page
+        Note over Worker: run ends
+    end
+    Queue->>DPS: UpsertIssueMessage
+    alt sourceModifiedDate unchanged and check enabled
+        Note over DPS: row skipped, nothing written
+    else
+        DPS->>DPS: create or update entity, flush
+    end
+```
+
+## Scheduled jobs
+
+Installed by the release playbook, see `.woodpecker/prod_itk_economics.yml`.
+
+| Job | Schedule | Command |
+| --- | --- | --- |
+| `sync-modified` | every 15 min | `app:data-providers:sync-modified` (window `PT1H`, async) |
+| `sync-deleted` | every 25 min | `app:data-providers:sync-deleted` (window `PT1H`, handled inline) |
+| `sync-deleted-week` | 02:50 | `app:data-providers:sync-deleted --interval=P1W` |
+| `full-sync-projects` | 02:00 | `app:data-providers:sync -j -p -d` |
+| `full-sync-workers` | 02:10 | `app:data-providers:sync -j -r -d` |
+| `full-sync-versions` | 02:20 | `app:data-providers:sync -j -s -d` |
+| `full-sync-issues` | 02:30 | `app:data-providers:sync -j -i -d` |
+| `full-sync-worklogs` | 02:40 | `app:data-providers:sync -j -w -d` |
+
+A single supervisor worker consumes the `async` transport — `APP_SUPERVISOR_WORKERS=1` and
+`messenger:consume --time-limit=900` in `docker-compose.server.override.yml`. That is why the full
+syncs are staggered ten minutes apart: a full run of one entity type takes a while, and nothing else
+drains the queue while it does.
+
+## Command options
+
+`app:data-providers:sync` selects what to sync and how:
+
+* `-j`, `--job`: dispatch to the `async` transport instead of handling everything inline.
+* `-a`, `--all`, or one of `-p` projects, `-s` versions, `-i` issues, `-w` worklogs, `-r` workers.
+* `--modified`: only fetch rows changed since the given date, passed on as `modifiedAfter`.
+* `-d`, `--disable-modified-at-check`: write every fetched row even when its modified timestamp is
+  unchanged. The nightly full syncs use this to repair rows that drifted out of sync.
+
+`sync-modified` and `sync-deleted` take `--interval` (a `DateInterval` string, default `PT1H`) and
+derive `modifiedAfter` / `deletedAfter` from it.
+
+## Details worth knowing
+
+* **Entity mapping.** Leantime milestones become Economics `Version`, tickets become `Issue`,
+  timesheets become `Worklog`, users become `Worker`.
+* **Project scoping.** Milestones, tickets and timesheets are requested only for the project ids
+  Economics already knows and includes (`ProjectRepository::getProjectTrackerIdsByDataProviders()`).
+  Projects and workers are fetched unscoped, so a new project has to be synced before its content
+  can follow.
+* **Paging.** On the entity endpoints `start` is an id cursor, not an offset: the next page starts at
+  the highest usable id on the page plus one. The delete endpoint pages on `deletionId` instead — the
+  deletion's own row id, not the deleted entity's. Deletions are ordered by when they happened while
+  the entity ids on a page are in no order at all, so paging on them would skip deletions. A page
+  shorter than the limit ends the run either way.
+* **Incrementality.** `modifiedAfter` filters on `itk_data_api_modified`, a column the plugin adds to
+  the Leantime tables and keeps current with triggers, because Leantime does not update its own
+  `modified` column on every write path.
+* **Deletions.** Leantime rows are gone by the time Economics asks, so the plugin records them in
+  `itk_projects_deleted`, `itk_tickets_deleted` and `itk_timesheets_deleted` via triggers, and the
+  `deleted` endpoint reads those tables. Three tables cover the four types because Leantime keeps
+  milestones and tickets in the same `zp_tickets` table, so both their deletions land in
+  `itk_tickets_deleted`. The endpoint serves one type per request: `deleteAsJob()` sends `type`,
+  `start`, `limit` and `deletedAfter`, and pages the way the entity endpoints do. The cursor advances
+  past a deletion that names no entity, because a skipped row still occupies a page position; a full
+  page with no usable `deletionId` stops the run with a logged error rather than re-queueing itself,
+  which would re-read the same page until the queue starves.
+* **Delete ordering.** The delete types run timesheets → tickets → milestones → projects, children
+  before the parents they hang off. What holds that order is the inline `sync://` transport rather
+  than the dispatch order: `deleteAll()` passes `asyncJobQueue` false, so every page of one type is
+  handled before the next type is dispatched. On the `async` transport the four types interleave, and
+  a project can be reached while its timesheets are still a page behind. That matters because a
+  parent which cannot be hard-deleted is only marked with `sourceDeletedDate`, and nothing revisits
+  the mark — so `sync-deleted` running inline is load-bearing, not an oversight to fix with `-j`.
+* **What blocks a removal.** `DataProviderService` hard-deletes an entity only when nothing points at
+  it. A project is kept if it still has invoices, issues, worklogs, versions, project billings or
+  service agreements; an issue is kept if it still has worklogs. Each of those points back with a
+  non-nullable, non-cascading foreign key, so removing anyway would be a database error rather than a
+  soft delete. Versions are always removable.
+* **First sync after a plugin install returns everything**, because installing stamps every existing
+  row with the install time.
+* **Failures.** A handler that fails with 408, 423, 425 or 429 rethrows, so the `async` transport
+  retries the message: three attempts spaced 10s, 30s and 90s, the last landing 130s after the first
+  failure. Any other 4xx describes the request itself, which no retry can change, so it becomes an
+  `UnrecoverableMessageHandlingException`, lands in the `failed` transport and is logged. Requests to
+  Leantime are capped at `timeout: 5` and `max_duration: 30` by `app.leantime.http_client`, so no
+  single page can hold the worker indefinitely.
+* **Retries are an `async` transport feature only.** `sync://` has no retry strategy, so the inline
+  `sync-deleted` run gets none: a failure there propagates out to the command.
+* **Authentication.** Each `DataProvider` row holds the Leantime base url and the API key sent as
+  `x-api-key`. Only providers with `class = App\Service\LeantimeApiService` and `enabled = true` are
+  synced.