fix(onec): complete outbound push pull and protect booked intervals
This commit is contained in:
Abdulkadyr Karagishiev
parent
33b0b89271
commit
b126f2c52d
@ -8,7 +8,7 @@
|
||||
|
||||
- **Frontend (`web/`):** React 18, Vite, Tailwind CSS, SSR рендеринг (`scripts/render-page.mjs`).
|
||||
- **Backend API (`cmd/server/`, `internal/`):** Go 1.22+ сервис обработки заявок, онлайн-записи, каталога услуг, врачей и чек-апов.
|
||||
- **Интеграция 1C (`bridge/bridge_1c.py`):** Python-мост для синхронизации слотов расписания, врачей, номенклатуры и создания заявок в 1С:Медицина.
|
||||
- **Интеграция 1C (`internal/api/onec*.go`, `internal/storage/onec*.go`):** автономный Push/Pull. 1С сама отправляет снимки и хуки, забирает pending и команды. `bridge/` сохранен как архив исходников; рабочий Compose его не запускает.
|
||||
- **База данных:** PostgreSQL 16 (данные сохраняются в Docker volume `dana-medical_db-data`).
|
||||
- **Медиафайлы:** сохраняются в volume `dana-medical_media` (монтируется в `/data/media`).
|
||||
|
||||
@ -48,7 +48,7 @@
|
||||
/home/mikail/dana-medical/deploy.sh
|
||||
```
|
||||
Скрипт автоматически:
|
||||
1. Затягивает свежие изменения из ветки `main` (`git fetch origin main && git reset --hard origin/main`).
|
||||
1. Затягивает свежие изменения из ветки `main` (`git fetch origin main && git merge --ff-only origin/main`).
|
||||
2. Собирает Docker-образ бэкенда и веб-части (`docker compose -p dana-medical -f docker-compose.prod.yml build api`).
|
||||
3. Бесшовно перезапускает контейнер API (`up -d api`).
|
||||
4. Проверяет `healthcheck` эндпоинта `/api/v1/health` и подтверждает успешность выкладки.
|
||||
|
||||
@ -80,10 +80,8 @@ func run() error {
|
||||
handler.WithClinicPBX("ochevidno", megafon.New(cfg.OchevidnoMegafonURL, cfg.OchevidnoMegafonKey))
|
||||
}
|
||||
go handler.RunPBXSync(ctx)
|
||||
handler.WithOneC(onec.New(cfg.OnecURL, cfg.OnecKey, cfg.OnecAuthHeader))
|
||||
handler.WithOneC(onec.New("", cfg.OnecKey, cfg.OnecAuthHeader))
|
||||
handler.WithOneCWebhook(cfg.OnecWebhookKey)
|
||||
handler.WithOneCTuning(time.Duration(cfg.OnecSyncSecs)*time.Second, cfg.OnecLongPoll)
|
||||
go handler.RunOneCSync(ctx)
|
||||
handler.WithBitrix(bitrix.New(cfg.BitrixWebhookURL))
|
||||
handler.WithCRM(crm.New(cfg.CrmAPIURL, cfg.CrmLeadIntakeSecret), cfg.CrmMode)
|
||||
go handler.RunBookingConfirm(ctx)
|
||||
|
||||
22
deploy.sh
22
deploy.sh
@ -9,22 +9,28 @@ cd "$DEPLOY_DIR"
|
||||
|
||||
# 1. Pull latest code from Gitea
|
||||
echo "Pulling latest changes from origin/main..."
|
||||
if [ -n "$(git status --porcelain)" ]; then
|
||||
echo "Deployment refused: working tree contains changes" >&2
|
||||
exit 1
|
||||
fi
|
||||
git fetch origin main
|
||||
git reset --hard origin/main
|
||||
git merge --ff-only origin/main
|
||||
|
||||
# 2. Build and restart API container
|
||||
echo "Building and restarting api service..."
|
||||
docker compose -p dana-medical -f docker-compose.prod.yml build api
|
||||
docker compose -p dana-medical -f docker-compose.prod.yml up -d api
|
||||
docker compose -p dana-medical -f docker-compose.prod.yml up -d --no-deps api
|
||||
|
||||
# 3. Check health status
|
||||
echo "Verifying health..."
|
||||
api_id="$(docker compose -p dana-medical -f docker-compose.prod.yml ps -q api)"
|
||||
test -n "$api_id"
|
||||
|
||||
healthy=false
|
||||
for i in $(seq 1 30); do
|
||||
status="$(docker inspect --format '{{.State.Health.Status}}' "$api_id" 2>/dev/null || echo "starting")"
|
||||
if [ "$status" = "healthy" ]; then
|
||||
healthy=true
|
||||
echo "API service is healthy!"
|
||||
break
|
||||
fi
|
||||
@ -36,4 +42,16 @@ for i in $(seq 1 30); do
|
||||
sleep 2
|
||||
done
|
||||
|
||||
if [ "$healthy" != true ]; then
|
||||
echo "API health check timed out" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# Outbound-only 1C polling supersedes the legacy bridge. Retain the stopped
|
||||
# container for rollback, but do not let it return on a Docker daemon restart.
|
||||
if docker inspect dana-1c-bridge >/dev/null 2>&1; then
|
||||
docker update --restart=no dana-1c-bridge >/dev/null
|
||||
docker stop --time 15 dana-1c-bridge >/dev/null
|
||||
fi
|
||||
|
||||
echo "[$(date --iso-8601=seconds)] Deployment completed successfully!"
|
||||
|
||||
@ -29,6 +29,9 @@ services:
|
||||
NODE_BINARY: node
|
||||
SITE_URL: ${SITE_URL:?SITE_URL is required}
|
||||
DEBUG: "false"
|
||||
ONEC_API_URL: ""
|
||||
ONEC_MCP_URL: ""
|
||||
ONEC_MCP_TOKEN: ""
|
||||
# Caddy уже работает в контейнере Ability Center. Публикуем DANA только
|
||||
# на docker0 хоста: из интернета порт недоступен, Caddy обращается к нему
|
||||
# как к 172.17.0.1:8082.
|
||||
@ -48,46 +51,6 @@ services:
|
||||
security_opt:
|
||||
- no-new-privileges:true
|
||||
|
||||
# Мост к 1C MCP. Единственный контейнер, которому разрешён egress на
|
||||
# 10.240.70.70:6003 (правило DOCKER-USER, см. deploy/firewall-1c-egress.sh).
|
||||
# IP закреплён: правило фаервола привязано к адресу, а не к имени.
|
||||
bridge:
|
||||
build:
|
||||
context: ./bridge
|
||||
image: dana-1c-bridge:latest
|
||||
container_name: dana-1c-bridge
|
||||
restart: unless-stopped
|
||||
env_file:
|
||||
- .env
|
||||
environment:
|
||||
ONEC_MCP_URL: ${ONEC_MCP_URL:-http://10.240.70.70:6003/api/execute_code}
|
||||
ONEC_MCP_TOKEN: ${ONEC_MCP_TOKEN:?ONEC_MCP_TOKEN is required}
|
||||
# Go-бэкенд авторизуется перед мостом тем же ключом, что шлёт в ONEC_API_URL.
|
||||
BRIDGE_API_KEY: ${ONEC_API_KEY:?ONEC_API_KEY is required}
|
||||
PG_HOST: db
|
||||
PG_PORT: "5432"
|
||||
PG_USER: dana
|
||||
PG_DB: dana
|
||||
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?POSTGRES_PASSWORD is required}
|
||||
depends_on:
|
||||
db:
|
||||
condition: service_healthy
|
||||
networks:
|
||||
default:
|
||||
ipv4_address: 172.20.0.50
|
||||
healthcheck:
|
||||
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8088/healthz', timeout=3)"]
|
||||
interval: 30s
|
||||
timeout: 5s
|
||||
retries: 5
|
||||
start_period: 10s
|
||||
security_opt:
|
||||
- no-new-privileges:true
|
||||
cap_drop:
|
||||
- ALL
|
||||
read_only: true
|
||||
tmpfs:
|
||||
- /tmp:size=16m
|
||||
|
||||
volumes:
|
||||
db-data:
|
||||
|
||||
35
docs/onec-autonomous.md
Normal file
35
docs/onec-autonomous.md
Normal file
@ -0,0 +1,35 @@
|
||||
# Автономная интеграция 1С
|
||||
|
||||
1С сама вызывает HTTPS API сайта; API не вызывает bridge или MCP. Запуск `RunOneCSync` удален, обратные сетевые вызовы создания, привязки и отмены из API удалены. Настройка `ONEC_WEBHOOK_KEY` включает автономный режим; Compose очищает старые `ONEC_API_URL`, `ONEC_MCP_URL`, `ONEC_MCP_TOKEN` в контейнере.
|
||||
|
||||
## Контракт
|
||||
|
||||
- `POST /api/v1/onec/sync/all`: полный снимок `branches`, `doctors`, `services`, `slots`, диапазон `from/to`. Пустое расписание очищает горизонт; успешный прием обновляет свежесть Push. При дубликатах слота `busy` имеет приоритет. Для совместимости снимок старого отправителя без границ принимает диапазон входящих слотов.
|
||||
- `GET /api/v1/onec/appointments/pending`: `data.appointments`, `data.commands`, `data.sync_requested`. Начало передается со смещением Москвы, включая комментарий и дату рождения, если они были указаны пользователем.
|
||||
- `POST /api/v1/onec/appointments/{id}/confirm`: UUID документа и окончание приема. Повтор с тем же UUID безопасен, несовместимое подтверждение/отказ после создания дает 409.
|
||||
- `POST /api/v1/onec/appointments/{id}/reject`: причина отказа, перевод в заявку оператору, если настроен CRM-контур.
|
||||
- `POST /api/v1/onec/events`: события статуса и запроса нового Push. Обратных HTTP-вызовов нет; устаревший `booked` не возвращает отмененную заявку. Изменение времени переносит интервал локальной занятости.
|
||||
- `POST /api/v1/onec/commands/{id}/ack`: `{success:true}` либо `{success:false,error:"..."}`. Неуспешная команда остается доступной. Типы команд: `link_bitrix`, `link_crm`, `cancel`. Отмена остается `cancel_pending` и занимает интервал до ACK 1С. После ACK просроченной записи отдельный повторяемый шаг закрывает лид Б24.
|
||||
|
||||
Все методы защищены ключом 1С. Привязка Б24 сохраняется долговечно и повторяется, даже если первый обмен после получения ID не удался. Создание самого лида Б24 остается существующим REST-клиентом; отдельная сделка не гарантируется одним наличием лида.
|
||||
|
||||
## Занятость и повторы
|
||||
|
||||
`BeginOnecAppointment` блокирует ключ запроса и врача через PostgreSQL advisory locks; выбор свободного окна и вставка pending выполняются в одной транзакции. Публичные свободные окна исключают пересечения с pending/booked/arrived по врачу, включая другие филиалы. Это правило действует и после следующего снимка, содержащего free. Заявки, отсеянные защитой от ботов, сразу сохраняются как failed для оператора и никогда не попадают в Pull-создание.
|
||||
|
||||
Миграция `0027_onec_autonomous.sql` добавляет комментарий/дату рождения в очередь, таблицы состояния Push и команд. Она не удаляет клинические данные и совместима со старым API при возврате образа. Перед применением нужен серверный pg_dump и сохраненный предыдущий образ.
|
||||
|
||||
## Приемка
|
||||
|
||||
Проверки запускаются на Linux в отдельной временной PostgreSQL:
|
||||
|
||||
```sh
|
||||
DANA_AUTONOMY_TEST_DSN=postgres://postgres:TEST_PASSWORD@127.0.0.1:5432/dana_autonomy_test?sslmode=disable go test -p 1 -count=1 ./...
|
||||
go vet ./...
|
||||
```
|
||||
|
||||
DSN должен указывать исключительно на временную тестовую базу. В сценариях проверяются публичный POST → pending, скрытие зарезервированного времени после Push, 12 конкурирующих запросов, повторный ключ, callbacks/409, запрос снимка, долговечные команды, ACK/повтор ACK и очистка пустого горизонта. Тесты не доказывают исполнение BSL на платформе 1С.
|
||||
|
||||
Обновление 1С: пять общих модулей `ДМ_СайтAPI`, `ДМ_СайтРасписание`, `ДМ_СайтЗапись`, `ДМ_СайтХуки`, `ДМ_СайтПодписки` из пакета в проекте config_dump. Нужны проверка на Test, обновление UMC и новый сквозной тест после загрузки. Резервный входящий HTTP-сервис 1С не нужен этому маршруту.
|
||||
|
||||
После проверки здоровья API `deploy.sh` останавливает только контейнер dana-1c-bridge и снимает его автозапуск. Сохраняются DB/media и общий Caddy. Откат: предыдущий образ и прежний Compose из серверного каталога rollback; bridge запускается вручную только если возвращен прежний режим.
|
||||
@ -239,7 +239,7 @@ const Onec = {
|
||||
}
|
||||
const STATUS = { booked: ['', 'Записан'], pending: ['wait', 'Отправляется'], failed: ['off', 'Не записан'], cancelled: ['grey', 'Отменён'], arrived: ['', 'Пришёл'] };
|
||||
const CONFIRM = {
|
||||
waiting: ['wait', 'Ждёт оператора'], confirmed: ['', 'Подтверждена'], cancelled: ['grey', 'Отменена оператором'],
|
||||
waiting: ['wait', 'Ждёт оператора'], cancel_pending: ['wait', 'Ожидает отмены в 1С'], confirmed: ['', 'Подтверждена'], cancelled: ['grey', 'Отменена оператором'],
|
||||
expired: ['off', 'Не подтвердили — отменена'], missed: ['grey', 'Приём без подтверждения'],
|
||||
lead_only: ['wait', 'Отправляется в Битрикс'], lead_sent: ['', 'Заявка в Битриксе'],
|
||||
};
|
||||
|
||||
@ -7,7 +7,6 @@ import (
|
||||
"net/http"
|
||||
"reflect"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"dana/internal/bitrix"
|
||||
"dana/internal/crm"
|
||||
@ -28,8 +27,6 @@ type API struct {
|
||||
pbxClinics map[string]*megafon.Client // АТС других клиник (clinics.go)
|
||||
onec *onec.Client
|
||||
onecHook string
|
||||
onecEvery time.Duration
|
||||
onecLongPoll bool
|
||||
bitrix *bitrix.Client
|
||||
crm *crm.Client
|
||||
crmMode string
|
||||
@ -141,6 +138,7 @@ func (a *API) Routes() http.Handler {
|
||||
mux.HandleFunc("GET /api/v1/onec/appointments/pending", a.onecPendingAppointments)
|
||||
mux.HandleFunc("POST /api/v1/onec/appointments/{id}/confirm", a.onecConfirmAppointment)
|
||||
mux.HandleFunc("POST /api/v1/onec/appointments/{id}/reject", a.onecRejectAppointment)
|
||||
mux.HandleFunc("POST /api/v1/onec/commands/{id}/ack", a.onecCommandAck)
|
||||
mux.HandleFunc("GET /api/v1/admin/onec", a.requireAuth(a.adminOnec))
|
||||
mux.HandleFunc("POST /api/v1/admin/onec/sync", a.requireAuth(a.adminOnecSync))
|
||||
mux.HandleFunc("PUT /api/v1/admin/onec/doctors/{onec_id}/link", a.requireAuth(a.adminOnecLink))
|
||||
|
||||
@ -13,7 +13,6 @@ import (
|
||||
|
||||
"dana/internal/bitrix"
|
||||
"dana/internal/crm"
|
||||
"dana/internal/onec"
|
||||
"dana/internal/storage"
|
||||
)
|
||||
|
||||
@ -83,7 +82,7 @@ func (a *API) bookingRiskReason(ctx context.Context, cfg storage.BookingSettings
|
||||
}
|
||||
|
||||
func (a *API) bookingConfirmEnabled() bool {
|
||||
return a.onec != nil && a.onec.Configured() && (a.bitrixEnabled() || a.crmEnabled())
|
||||
return a.onecAutonomousEnabled() && (a.bitrixEnabled() || a.crmEnabled())
|
||||
}
|
||||
|
||||
// kickConfirm будит фоновую проверку сразу после новой записи, чтобы лид появился без ожидания минуты.
|
||||
@ -157,6 +156,22 @@ func (a *API) RunBookingConfirm(ctx context.Context) {
|
||||
}
|
||||
|
||||
func (a *API) processConfirmQueue(ctx context.Context, now time.Time) error {
|
||||
if a.bitrixEnabled() {
|
||||
expired, err := a.store.ExpiredOnecLeads(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, v := range expired {
|
||||
if err = a.bitrix.FailLead(ctx, v.LeadID); err != nil {
|
||||
slog.Warn("expire Bitrix lead after 1C ACK", "lead", v.LeadID, "err", err)
|
||||
continue
|
||||
}
|
||||
if err = a.store.MarkExpiredOnecLead(ctx, v.CommandID); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if err := a.sendLeadOnly(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
@ -188,13 +203,6 @@ func (a *API) processConfirmQueue(ctx context.Context, now time.Time) error {
|
||||
it.BitrixLeadID = id
|
||||
slog.Info("booking confirm: лид создан в Битрикс", "appointment", it.ID, "lead", id)
|
||||
|
||||
if it.OnecID != "" {
|
||||
if err := a.onec.LinkBitrix(ctx, it.OnecID, id); err != nil {
|
||||
slog.Warn("booking confirm: не удалось связать 1С с Битриксом", "onec_id", it.OnecID, "lead", id, "err", err)
|
||||
} else {
|
||||
slog.Info("booking confirm: заявка в 1С связана с лидом Битрикса", "onec_id", it.OnecID, "lead", id)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -211,15 +219,27 @@ func (a *API) processConfirmQueue(ctx context.Context, now time.Time) error {
|
||||
slog.Info("booking confirm: сделка создана в CRM", "appointment", it.ID, "deal", dealID)
|
||||
|
||||
if it.OnecID != "" {
|
||||
if err := a.onec.LinkCRM(ctx, it.OnecID, dealID); err != nil {
|
||||
if err := a.store.QueueOnecCommand(ctx, it.ID, "link_crm", dealID, ""); err != nil {
|
||||
slog.Warn("booking confirm: не удалось связать 1С со сделкой CRM", "onec_id", it.OnecID, "deal", dealID, "err", err)
|
||||
} else {
|
||||
slog.Info("booking confirm: заявка в 1С связана со сделкой CRM", "onec_id", it.OnecID, "deal", dealID)
|
||||
slog.Info("booking confirm: связь 1С с CRM поставлена в очередь", "onec_id", it.OnecID, "deal", dealID)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if it.OnecID != "" {
|
||||
if it.BitrixLeadID != "" {
|
||||
if err := a.store.QueueOnecCommand(ctx, it.ID, "link_bitrix", it.BitrixLeadID, ""); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if it.CrmDealID != "" {
|
||||
if err := a.store.QueueOnecCommand(ctx, it.ID, "link_crm", it.CrmDealID, ""); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
if it.BitrixLeadID != "" {
|
||||
bitrixIds = append(bitrixIds, it.BitrixLeadID)
|
||||
}
|
||||
@ -241,28 +261,16 @@ func (a *API) processConfirmQueue(ctx context.Context, now time.Time) error {
|
||||
case confirmMissed:
|
||||
a.finishConfirm(ctx, it, "missed")
|
||||
case confirmCancel:
|
||||
if a.cancelUnconfirmed(ctx, it) {
|
||||
a.finishConfirm(ctx, it, "cancelled")
|
||||
a.leadNote(ctx, it, "Сайт отменил запись в 1С: лид переведён в провальную стадию. Окно у врача снова свободно.")
|
||||
}
|
||||
a.queueAutonomousCancel(ctx, it, "cancelled")
|
||||
case confirmExpire:
|
||||
if a.cancelUnconfirmed(ctx, it) {
|
||||
a.finishConfirm(ctx, it, "expired")
|
||||
if err := a.bitrix.FailLead(ctx, it.BitrixLeadID); err != nil {
|
||||
slog.Warn("booking confirm: стадия лида не обновлена", "lead", it.BitrixLeadID, "err", err)
|
||||
}
|
||||
a.leadNote(ctx, it, "Запись не подтвердили вовремя — сайт отменил её в 1С и перевёл лид в «Клиент не записался». "+
|
||||
"Если пациент всё же придёт, запишите его заново.")
|
||||
}
|
||||
a.queueAutonomousCancel(ctx, it, "expired")
|
||||
}
|
||||
}
|
||||
} else if !a.bitrixEnabled() && a.crmEnabled() {
|
||||
// В режиме CRM-only проверяем автоотмену по истечении срока подтверждения
|
||||
for _, it := range items {
|
||||
if d, ok := confirmDeadline(it.CreatedAt, it.Start, cfg); ok && !now.Before(d) {
|
||||
if a.cancelUnconfirmed(ctx, it) {
|
||||
a.finishConfirm(ctx, it, "expired")
|
||||
}
|
||||
a.queueAutonomousCancel(ctx, it, "expired")
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -325,26 +333,10 @@ func (a *API) finishConfirm(ctx context.Context, it storage.ConfirmItem, state s
|
||||
slog.Info("booking confirm", "appointment", it.ID, "lead", it.BitrixLeadID, "state", state)
|
||||
}
|
||||
|
||||
// cancelUnconfirmed отменяет запись в 1С. false — 1С не ответила, попробуем на следующем круге.
|
||||
func (a *API) cancelUnconfirmed(ctx context.Context, it storage.ConfirmItem) bool {
|
||||
if it.OnecID != "" {
|
||||
err := a.onec.Cancel(ctx, it.OnecID)
|
||||
// 404 1С возвращает как «не найден»: записи там уже нет — значит, и отменять нечего
|
||||
if err != nil && !errors.Is(err, onec.ErrDoctorNotFound) {
|
||||
slog.Warn("booking confirm: 1С не отменила запись", "appointment", it.ID, "err", err)
|
||||
_ = a.store.SetConfirmError(ctx, it.ID, "1С: "+onecErrText(err))
|
||||
return false
|
||||
}
|
||||
}
|
||||
if err := a.resyncOnecDoctor(ctx, it.DoctorOnec); err != nil {
|
||||
slog.Warn("booking confirm: окна врача не обновились", "doctor", it.DoctorOnec, "err", err)
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (a *API) leadNote(ctx context.Context, it storage.ConfirmItem, text string) {
|
||||
if err := a.bitrix.AddLeadComment(ctx, it.BitrixLeadID, text); err != nil {
|
||||
slog.Warn("booking confirm: комментарий в лид не добавлен", "lead", it.BitrixLeadID, "err", err)
|
||||
// Cancellation remains occupied until the polling 1C session acknowledges its durable command.
|
||||
func (a *API) queueAutonomousCancel(ctx context.Context, it storage.ConfirmItem, finalState string) {
|
||||
if err := a.store.QueueOnecCommand(ctx, it.ID, "cancel", "", finalState); err != nil {
|
||||
slog.Warn("queue autonomous cancellation", "appointment", it.ID, "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -10,7 +10,6 @@ import (
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
|
||||
@ -19,98 +18,23 @@ import (
|
||||
"dana/internal/storage"
|
||||
)
|
||||
|
||||
// Расписание врачей и запись на приём из 1С: окна забираются по таймеру и хранятся у нас,
|
||||
// перед записью окно перепроверяется в 1С. Если 1С не отвечает, сайт не падает, а принимает обычную заявку.
|
||||
// 1С сама отправляет расписание и забирает pending. Сайт не вызывает 1С по сети.
|
||||
|
||||
const (
|
||||
onecHorizonDays = 14
|
||||
onecSyncEvery = 5 * time.Minute
|
||||
onecDirEvery = time.Hour
|
||||
onecStaleAfter = 24 * time.Hour // после этого срока без удачной синхронизации окна не показываем
|
||||
onecSourceLabel = "Сайт"
|
||||
maxAppointmentIn = 16 << 10
|
||||
)
|
||||
|
||||
// WithOneC подключает 1С; без адреса или ключа запись через 1С выключена.
|
||||
func (a *API) WithOneC(c *onec.Client) { a.onec = c }
|
||||
|
||||
// WithOneCTuning задаёт период опроса (0 — по умолчанию) и включает long polling через GET /changes.
|
||||
func (a *API) WithOneCTuning(every time.Duration, longPoll bool) {
|
||||
a.onecEvery, a.onecLongPoll = every, longPoll
|
||||
}
|
||||
|
||||
var onecSyncMu sync.Mutex // одна синхронизация за раз: фоновая и ручная не пересекаются
|
||||
|
||||
func onecWindow() (from, to time.Time) {
|
||||
now := time.Now().In(moscow)
|
||||
from = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, moscow)
|
||||
return from, from.AddDate(0, 0, onecHorizonDays)
|
||||
}
|
||||
|
||||
// syncOneC забирает расписание на 14 дней; справочники (филиалы, врачи, услуги) — при dir=true.
|
||||
func (a *API) syncOneC(ctx context.Context, dir bool) (err error) {
|
||||
if !a.onec.Configured() {
|
||||
return errors.New("1С не подключена: задайте ONEC_API_URL и ONEC_API_KEY")
|
||||
}
|
||||
onecSyncMu.Lock()
|
||||
defer onecSyncMu.Unlock()
|
||||
logID, err := a.store.StartOnecSync(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
doctors, slotsN := 0, 0
|
||||
defer func() {
|
||||
errText := ""
|
||||
if err != nil {
|
||||
errText = onecErrText(err)
|
||||
}
|
||||
if e := a.store.FinishOnecSync(context.Background(), logID, doctors, slotsN, errText); e != nil {
|
||||
slog.Warn("onec sync log", "err", e)
|
||||
}
|
||||
}()
|
||||
|
||||
if dir {
|
||||
branches, err := a.onec.Branches(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("филиалы: %w", err)
|
||||
}
|
||||
docs, err := a.onec.Doctors(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("врачи: %w", err)
|
||||
}
|
||||
linked, err := a.store.SaveOnecDirectory(ctx, branches, docs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
doctors = len(docs)
|
||||
if linked > 0 {
|
||||
slog.Info("onec: врачи привязаны по ФИО", "count", linked)
|
||||
}
|
||||
ids, err := a.store.LinkedOnecDoctorIDs(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, id := range ids {
|
||||
svc, err := a.onec.Services(ctx, id)
|
||||
if err != nil {
|
||||
slog.Warn("onec: услуги врача", "doctor", id, "err", err)
|
||||
continue
|
||||
}
|
||||
if err := a.store.SaveOnecServices(ctx, id, svc); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
from, to := onecWindow()
|
||||
slots, err := a.onec.Schedule(ctx, from, to, "")
|
||||
if err != nil {
|
||||
return fmt.Errorf("расписание: %w", err)
|
||||
}
|
||||
slotsN = len(slots)
|
||||
return a.store.ReplaceOnecSlots(ctx, from, to, slots)
|
||||
}
|
||||
|
||||
func onecErrText(err error) string {
|
||||
if errors.Is(err, onec.ErrUnauthorized) {
|
||||
return "1С не приняла ключ доступа: проверьте ONEC_API_KEY и что IP сервера сайта открыт в 1С. " + err.Error()
|
||||
@ -118,45 +42,6 @@ func onecErrText(err error) string {
|
||||
return err.Error()
|
||||
}
|
||||
|
||||
// RunOneCSync — фоновая синхронизация: при старте и каждые 5 минут; справочники — раз в час.
|
||||
func (a *API) RunOneCSync(ctx context.Context) {
|
||||
if !a.onec.Configured() {
|
||||
return
|
||||
}
|
||||
slog.Info("onec sync enabled", "host", a.onec.Host())
|
||||
lastDir := time.Time{}
|
||||
run := func() {
|
||||
dir := time.Since(lastDir) >= onecDirEvery
|
||||
c, cancel := context.WithTimeout(ctx, 2*time.Minute)
|
||||
defer cancel()
|
||||
if err := a.syncOneC(c, dir); err != nil {
|
||||
slog.Warn("onec sync", "err", err)
|
||||
return
|
||||
}
|
||||
if dir {
|
||||
lastDir = time.Now()
|
||||
}
|
||||
}
|
||||
run()
|
||||
every := a.onecEvery
|
||||
if every <= 0 {
|
||||
every = onecSyncEvery
|
||||
}
|
||||
if a.onecLongPoll {
|
||||
go a.runOneCLongPoll(ctx)
|
||||
}
|
||||
t := time.NewTicker(every)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
run()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- Публичное API ----------
|
||||
|
||||
// doctorSlots отдаёт свободные окна врача и услуги приёма. enabled=false — запись через 1С недоступна,
|
||||
@ -164,7 +49,7 @@ func (a *API) RunOneCSync(ctx context.Context) {
|
||||
func (a *API) doctorSlots(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Cache-Control", "no-store")
|
||||
off := func() { ok(w, map[string]any{"enabled": false, "slots": []any{}, "services": []any{}}, nil) }
|
||||
if !a.onec.Configured() {
|
||||
if !a.onecAutonomousEnabled() {
|
||||
off()
|
||||
return
|
||||
}
|
||||
@ -258,11 +143,15 @@ func (a *API) createAppointment(w http.ResponseWriter, r *http.Request) {
|
||||
fail(w, http.StatusBadRequest, "start_invalid", "Выберите время приёма ещё раз")
|
||||
return
|
||||
}
|
||||
if !a.onec.Configured() {
|
||||
if !a.onecAutonomousEnabled() {
|
||||
fail(w, http.StatusServiceUnavailable, "onec_disabled", "Онлайн-запись сейчас недоступна, оставьте заявку")
|
||||
return
|
||||
}
|
||||
ctx := r.Context()
|
||||
if !a.store.OnecFresh(ctx, onecStaleAfter) {
|
||||
fail(w, 503, "onec_stale", "Расписание временно недоступно")
|
||||
return
|
||||
}
|
||||
onecID, doctorID, err := a.store.OnecDoctorBySlug(ctx, in.DoctorSlug)
|
||||
if err != nil {
|
||||
handleErr(w, err)
|
||||
@ -306,10 +195,18 @@ func (a *API) createAppointment(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
initialStatus := "pending"
|
||||
if leadOnly != "" {
|
||||
initialStatus = "failed"
|
||||
}
|
||||
rec, created, err := a.store.BeginOnecAppointment(ctx, storage.OnecAppointment{
|
||||
RequestID: in.RequestID, DoctorOnec: onecID, BranchID: in.BranchID, ServiceID: in.ServiceID,
|
||||
Start: start, Patient: in.Name, Phone: in.Phone,
|
||||
Start: start, Patient: in.Name, Phone: in.Phone, Comment: in.Comment, BirthDate: in.BirthDate, Status: initialStatus,
|
||||
})
|
||||
if errors.Is(err, storage.ErrOnecSlotBusy) {
|
||||
fail(w, http.StatusConflict, "slot_busy", "Окно уже занято; выберите другое время")
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
handleErr(w, err)
|
||||
return
|
||||
@ -325,108 +222,21 @@ func (a *API) createAppointment(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
var slot *onec.Slot
|
||||
var end time.Time
|
||||
var apptID string
|
||||
|
||||
if a.onec.Configured() {
|
||||
// Окно ещё свободно? Спрашиваем 1С по этому врачу на этот день.
|
||||
dayFrom := time.Date(start.In(moscow).Year(), start.In(moscow).Month(), start.In(moscow).Day(), 0, 0, 0, 0, moscow)
|
||||
dayTo := dayFrom.AddDate(0, 0, 1)
|
||||
fresh, err := a.onec.Schedule(ctx, dayFrom, dayTo, onecID)
|
||||
if err != nil {
|
||||
a.appointmentFallback(w, r, in, doctorID, rec, err)
|
||||
return
|
||||
}
|
||||
_ = a.store.ReplaceOnecDoctorSlots(ctx, onecID, dayFrom, dayTo, fresh)
|
||||
for i := range fresh {
|
||||
s := fresh[i]
|
||||
if s.Free() && s.Start.Equal(start) && (in.BranchID == "" || s.BranchID == in.BranchID) {
|
||||
slot = &s
|
||||
break
|
||||
}
|
||||
}
|
||||
if slot == nil {
|
||||
a.slotTaken(w, r, rec, onecID, "окно уже занято")
|
||||
return
|
||||
}
|
||||
|
||||
appt, err := a.onec.Book(ctx, onec.AppointmentRequest{
|
||||
RequestID: in.RequestID, DoctorID: onecID, BranchID: slot.BranchID, ServiceID: in.ServiceID, SlotID: slot.SlotID,
|
||||
Start: slot.Start, Name: in.Name, Phone: in.Phone, BirthDate: in.BirthDate, Comment: in.Comment, Source: onecSourceLabel,
|
||||
branch := a.store.OnecBranchName(ctx, rec.BranchID)
|
||||
leadID := a.saveAppointmentLead(ctx, r, in, doctorID, fmt.Sprintf("Ожидает регистрации в 1С: %s, %s.%s",
|
||||
formatSlotTime(rec.Start), branchOrDash(branch), a.serviceNote(ctx, onecID, in.ServiceID)), map[string]any{
|
||||
"onec_doctor_id": onecID, "slot_start": rec.Start.Format(time.RFC3339),
|
||||
})
|
||||
switch {
|
||||
case errors.Is(err, onec.ErrSlotBusy):
|
||||
_ = a.store.MarkOnecSlotBusy(ctx, onecID, slot.BranchID, slot.Start)
|
||||
a.slotTaken(w, r, rec, onecID, err.Error())
|
||||
return
|
||||
case err != nil:
|
||||
a.appointmentFallback(w, r, in, doctorID, rec, err)
|
||||
return
|
||||
}
|
||||
_ = a.store.MarkOnecSlotBusy(ctx, onecID, slot.BranchID, slot.Start)
|
||||
|
||||
end = appt.End
|
||||
if end.IsZero() {
|
||||
end = slot.End
|
||||
}
|
||||
apptID = appt.ID
|
||||
} else {
|
||||
// Режим Push/Pull (1С сама опрашивает очередь по регламентному заданию)
|
||||
localSlots, err := a.store.FreeOnecSlots(ctx, onecID, start, start.Add(time.Minute))
|
||||
if err != nil || len(localSlots) == 0 {
|
||||
a.slotTaken(w, r, rec, onecID, "окно уже занято")
|
||||
return
|
||||
}
|
||||
for i := range localSlots {
|
||||
ls := localSlots[i]
|
||||
if in.BranchID == "" || ls.BranchID == in.BranchID {
|
||||
slot = &onec.Slot{
|
||||
DoctorID: onecID,
|
||||
BranchID: ls.BranchID,
|
||||
Start: ls.Start,
|
||||
End: ls.End,
|
||||
Status: "free",
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
if slot == nil {
|
||||
a.slotTaken(w, r, rec, onecID, "окно уже занято")
|
||||
return
|
||||
}
|
||||
_ = a.store.MarkOnecSlotBusy(ctx, onecID, slot.BranchID, slot.Start)
|
||||
end = slot.End
|
||||
}
|
||||
|
||||
branch := a.store.OnecBranchName(ctx, slot.BranchID)
|
||||
confirm := a.bookingConfirmEnabled()
|
||||
note := "Запись подтверждена в 1С"
|
||||
if confirm {
|
||||
note = "Запись сделана в 1С, ждёт подтверждения оператором в Битрикс24"
|
||||
}
|
||||
leadID := a.saveAppointmentLead(ctx, r, in, doctorID, fmt.Sprintf("%s: %s, %s.%s",
|
||||
note, formatSlotTime(slot.Start), branchOrDash(branch), a.serviceNote(ctx, onecID, in.ServiceID)), map[string]any{
|
||||
"onec_appointment_id": apptID, "onec_doctor_id": onecID, "slot_start": slot.Start.Format(time.RFC3339),
|
||||
})
|
||||
if apptID != "" {
|
||||
if err := a.store.FinishOnecAppointment(context.Background(), rec.ID, "booked", apptID, "", &end, leadID); err != nil {
|
||||
slog.Error("onec appointment finish", "id", rec.ID, "err", err)
|
||||
}
|
||||
if confirm {
|
||||
if err := a.store.StartBookingConfirm(context.Background(), rec.ID); err != nil {
|
||||
slog.Error("booking confirm start", "id", rec.ID, "err", err)
|
||||
}
|
||||
a.kickConfirm()
|
||||
}
|
||||
if err := a.store.AttachOnecLead(ctx, rec.ID, leadID); err != nil {
|
||||
slog.Error("attach pending appointment lead", "id", rec.ID, "err", err)
|
||||
}
|
||||
wcm.consume(in.VerificationID)
|
||||
writeJSON(w, http.StatusCreated, envelope{Data: map[string]any{
|
||||
"mode": "booked",
|
||||
"start": slot.Start,
|
||||
"end": end,
|
||||
"mode": "pending",
|
||||
"start": rec.Start,
|
||||
"end": rec.End,
|
||||
"branch": branch,
|
||||
"message": a.bookedMessage(slot.Start),
|
||||
"message": "Заявка принята. Подтверждение записи ожидается от клиники.",
|
||||
}})
|
||||
}
|
||||
|
||||
@ -441,6 +251,8 @@ func (a *API) bookedMessage(start time.Time) string {
|
||||
// repeatAppointment отвечает на повтор того же запроса (тот же request_id), не создавая записи заново.
|
||||
func (a *API) repeatAppointment(w http.ResponseWriter, rec storage.OnecAppointment) {
|
||||
switch rec.Status {
|
||||
case "pending":
|
||||
writeJSON(w, http.StatusAccepted, envelope{Data: map[string]any{"mode": "pending", "start": rec.Start, "message": "Заявка принята. Ожидается подтверждение клиники."}})
|
||||
case "booked":
|
||||
writeJSON(w, http.StatusOK, envelope{Data: map[string]any{
|
||||
"mode": "booked", "start": rec.Start, "message": a.bookedMessage(rec.Start),
|
||||
@ -546,18 +358,16 @@ func (a *API) adminOnec(w http.ResponseWriter, r *http.Request, u *storage.Admin
|
||||
return
|
||||
}
|
||||
ok(w, map[string]any{
|
||||
"configured": a.onec.Configured(),
|
||||
"host": a.onec.Host(),
|
||||
"configured": a.onecAutonomousEnabled(),
|
||||
"host": "Push/Pull",
|
||||
"status": st,
|
||||
"doctors": doctors,
|
||||
}, nil)
|
||||
}
|
||||
|
||||
func (a *API) adminOnecSync(w http.ResponseWriter, r *http.Request, u *storage.AdminUser) {
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 2*time.Minute)
|
||||
defer cancel()
|
||||
if err := a.syncOneC(ctx, true); err != nil {
|
||||
fail(w, http.StatusBadGateway, "onec_sync", onecErrText(err))
|
||||
if err := a.store.RequestOnecSnapshot(r.Context()); err != nil {
|
||||
handleErr(w, err)
|
||||
return
|
||||
}
|
||||
a.adminOnec(w, r, u)
|
||||
@ -636,44 +446,3 @@ func (a *API) adminOnecAppointments(w http.ResponseWriter, r *http.Request, u *s
|
||||
a.store.Audit(r.Context(), u.ID, "view", "onec_appointments", 0, clientIP(r))
|
||||
ok(w, list, map[string]any{"total": len(list)})
|
||||
}
|
||||
|
||||
// runOneCLongPoll держит у 1С запрос /changes: как только версия расписания выросла, окна перечитываются сразу.
|
||||
func (a *API) runOneCLongPoll(ctx context.Context) {
|
||||
slog.Info("onec long polling enabled")
|
||||
var since int64
|
||||
first := true
|
||||
for ctx.Err() == nil {
|
||||
wait := 25 * time.Second
|
||||
if first {
|
||||
wait = 0 // первый запрос только узнаёт текущую версию, не ожидая изменений
|
||||
}
|
||||
version, ids, err := a.onec.Changes(ctx, since, wait)
|
||||
if err != nil {
|
||||
slog.Warn("onec long poll", "err", err)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(5 * time.Second):
|
||||
}
|
||||
continue
|
||||
}
|
||||
if first {
|
||||
first = false
|
||||
since = version // первый ответ только запоминает версию: полный опрос уже сделан при старте
|
||||
continue
|
||||
}
|
||||
if version <= since {
|
||||
continue
|
||||
}
|
||||
since = version
|
||||
if len(ids) == 0 {
|
||||
a.onecFullResyncAsync(false)
|
||||
continue
|
||||
}
|
||||
for _, id := range ids {
|
||||
if err := a.resyncOnecDoctor(ctx, id); err != nil {
|
||||
slog.Warn("onec long poll: врач не обновился", "doctor", id, "err", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
142
internal/api/onec_autonomous_test.go
Normal file
142
internal/api/onec_autonomous_test.go
Normal file
@ -0,0 +1,142 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"dana/internal/onec"
|
||||
"dana/internal/storage"
|
||||
)
|
||||
|
||||
func TestAutonomousHTTPHandshake(t *testing.T) {
|
||||
dsn := os.Getenv("DANA_AUTONOMY_TEST_DSN")
|
||||
if dsn == "" {
|
||||
t.Skip("requires disposable autonomy DB")
|
||||
}
|
||||
ctx := context.Background()
|
||||
st, err := storage.New(ctx, dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer st.Close()
|
||||
if err = st.Migrate(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
app := &API{store: st, limiter: newRateLimiter(), onecHook: "autonomy-test"}
|
||||
mux := app.Routes()
|
||||
call := func(method, path string, body any, want int) *httptest.ResponseRecorder {
|
||||
t.Helper()
|
||||
b, _ := json.Marshal(body)
|
||||
req := httptest.NewRequest(method, path, bytes.NewReader(b))
|
||||
req.Header.Set("X-Api-Key", "autonomy-test")
|
||||
w := httptest.NewRecorder()
|
||||
mux.ServeHTTP(w, req)
|
||||
if w.Code != want {
|
||||
t.Fatalf("%s %s: %d, want %d: %s", method, path, w.Code, want, w.Body.String())
|
||||
}
|
||||
return w
|
||||
}
|
||||
start := time.Now().UTC().Truncate(time.Hour).Add(72 * time.Hour)
|
||||
end := start.Add(30 * time.Minute)
|
||||
from, to := start.Add(-time.Hour), end.Add(time.Hour)
|
||||
doctor := fmt.Sprintf("http-autonomy-%d", time.Now().UnixNano())
|
||||
slot := onec.Slot{DoctorID: doctor, BranchID: "branch", Start: start, End: end, Status: "free"}
|
||||
payload := onecSyncPayload{From: &from, To: &to, Branches: []onec.Branch{}, Doctors: []onec.Doctor{}, Services: []onecDoctorServices{}, Slots: []onec.Slot{slot}}
|
||||
call("POST", "/api/v1/onec/sync/all", payload, 200)
|
||||
call("POST", "/api/v1/onec/sync/all", map[string]any{}, 400)
|
||||
var doctorID int64
|
||||
if err = st.Pool().QueryRow(ctx, `INSERT INTO doctors(slug,last_name,first_name,booking_provider) VALUES($1,'Тест','Врач','native') RETURNING id`, doctor).Scan(&doctorID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err = st.Pool().Exec(ctx, `INSERT INTO onec_doctors(onec_id,full_name,doctor_id) VALUES($1,'Тест Врач',$2)`, doctor, doctorID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = st.SetBookingSettings(ctx, storage.BookingSettings{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
call("GET", "/api/v1/doctors/"+doctor+"/slots", nil, 200)
|
||||
creation := appointmentInput{DoctorSlug: doctor, Start: start.Format(time.RFC3339), Name: "Тестовый пациент", Phone: "79990000000", Consent: true, RequestID: doctor, Comment: "HTTP test"}
|
||||
made := call("POST", "/api/v1/appointments", creation, 201)
|
||||
if !bytes.Contains(made.Body.Bytes(), []byte(`"mode":"pending"`)) {
|
||||
t.Fatal("unconfirmed booking reported as booked")
|
||||
}
|
||||
call("POST", "/api/v1/appointments", creation, 202)
|
||||
pendingRecords, err := st.PendingOnecAppointments(ctx, 50)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var rec storage.OnecPendingAppointment
|
||||
for _, v := range pendingRecords {
|
||||
if v.RequestID == doctor {
|
||||
rec = v
|
||||
}
|
||||
}
|
||||
if rec.ID == 0 {
|
||||
t.Fatal("public booking not queued")
|
||||
}
|
||||
call("POST", "/api/v1/onec/sync/all", payload, 200)
|
||||
free := call("GET", "/api/v1/doctors/"+doctor+"/slots", nil, 200)
|
||||
var freeBody struct {
|
||||
Data struct {
|
||||
Enabled bool `json:"enabled"`
|
||||
Slots []storage.OnecSlot `json:"slots"`
|
||||
} `json:"data"`
|
||||
}
|
||||
if err = json.Unmarshal(free.Body.Bytes(), &freeBody); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !freeBody.Data.Enabled || len(freeBody.Data.Slots) != 0 {
|
||||
t.Fatal("push reopened reserved public slot")
|
||||
}
|
||||
|
||||
pending := call("GET", "/api/v1/onec/appointments/pending", nil, 200)
|
||||
if !bytes.Contains(pending.Body.Bytes(), []byte(doctor)) || !bytes.Contains(pending.Body.Bytes(), []byte(`"commands"`)) {
|
||||
t.Fatal("pending contract incomplete")
|
||||
}
|
||||
uuid := "00000000-0000-0000-0000-000000000002"
|
||||
confirm := confirmAppointmentReq{OnecAppointmentID: uuid, End: &end}
|
||||
path := fmt.Sprintf("/api/v1/onec/appointments/%d/confirm", rec.ID)
|
||||
call("POST", path, confirm, 200)
|
||||
call("POST", path, confirm, 200)
|
||||
confirm.OnecAppointmentID = "00000000-0000-0000-0000-000000000003"
|
||||
call("POST", path, confirm, 409)
|
||||
call("POST", fmt.Sprintf("/api/v1/onec/appointments/%d/reject", rec.ID), rejectAppointmentReq{Reason: "late"}, 409)
|
||||
call("POST", "/api/v1/onec/events", onecEvent{Event: "schedule_changed", DoctorID: doctor}, 202)
|
||||
pending = call("GET", "/api/v1/onec/appointments/pending", nil, 200)
|
||||
if !bytes.Contains(pending.Body.Bytes(), []byte(`"sync_requested":true`)) {
|
||||
t.Fatal("snapshot request not exposed")
|
||||
}
|
||||
if err = st.QueueOnecCommand(ctx, rec.ID, "cancel", "", "cancelled"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
commands, err := st.PendingOnecCommands(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var id int64
|
||||
for _, c := range commands {
|
||||
if c.AppointmentID == uuid {
|
||||
id = c.ID
|
||||
}
|
||||
}
|
||||
if id == 0 {
|
||||
t.Fatal("cancel command missing")
|
||||
}
|
||||
ack := fmt.Sprintf("/api/v1/onec/commands/%d/ack", id)
|
||||
call("POST", ack, map[string]any{"success": true}, 200)
|
||||
call("POST", ack, map[string]any{"success": true}, 200)
|
||||
call("POST", "/api/v1/onec/events", onecEvent{Event: "appointment_status", AppointmentID: uuid, Status: "cancelled"}, 200)
|
||||
call("POST", path, confirm, 409)
|
||||
payload.Slots = []onec.Slot{}
|
||||
call("POST", "/api/v1/onec/sync/all", payload, 200)
|
||||
slots, err := st.FreeOnecSlots(ctx, doctor, from, to)
|
||||
if err != nil || len(slots) != 0 {
|
||||
t.Fatal("empty horizon did not clear slots")
|
||||
}
|
||||
}
|
||||
@ -1,8 +1,11 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"dana/internal/storage"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
@ -29,7 +32,11 @@ func (a *API) checkOnecKey(r *http.Request) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
var uuidTextRE = regexp.MustCompile(`^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$`)
|
||||
|
||||
type onecSyncPayload struct {
|
||||
From *time.Time `json:"from"`
|
||||
To *time.Time `json:"to"`
|
||||
Branches []onec.Branch `json:"branches"`
|
||||
Doctors []onec.Doctor `json:"doctors"`
|
||||
Services []onecDoctorServices `json:"services"`
|
||||
@ -54,6 +61,28 @@ func (a *API) onecSyncAll(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
if payload.Branches == nil || payload.Doctors == nil || payload.Services == nil || payload.Slots == nil || (payload.From == nil) != (payload.To == nil) {
|
||||
fail(w, 400, "validation", "Нужен полный снимок branches/doctors/services/slots и согласованный диапазон from/to")
|
||||
return
|
||||
}
|
||||
payload.Slots = mergeOnecSlots(payload.Slots)
|
||||
from, to := onecWindow()
|
||||
if payload.From != nil && payload.To != nil {
|
||||
from, to = *payload.From, *payload.To
|
||||
} else if len(payload.Slots) > 0 {
|
||||
from, to = findSlotBounds(payload.Slots)
|
||||
}
|
||||
if !from.Before(to) || to.Sub(from) > 32*24*time.Hour {
|
||||
fail(w, 400, "validation", "Некорректный диапазон снимка")
|
||||
return
|
||||
}
|
||||
for _, sl := range payload.Slots {
|
||||
if sl.DoctorID == "" || !sl.Start.Before(sl.End) || sl.Start.Before(from) || !sl.Start.Before(to) || (sl.Status != "free" && sl.Status != "busy") {
|
||||
fail(w, 400, "validation", "Некорректный слот")
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
ctx := r.Context()
|
||||
linked := 0
|
||||
var err error
|
||||
@ -69,7 +98,7 @@ func (a *API) onecSyncAll(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
// 2. Сохраняем услуги врачей
|
||||
for _, ds := range payload.Services {
|
||||
if ds.DoctorID != "" && len(ds.Services) > 0 {
|
||||
if ds.DoctorID != "" {
|
||||
if err := a.store.SaveOnecServices(ctx, ds.DoctorID, ds.Services); err != nil {
|
||||
fail(w, http.StatusInternalServerError, "db_error", "Ошибка сохранения услуг: "+err.Error())
|
||||
return
|
||||
@ -77,13 +106,13 @@ func (a *API) onecSyncAll(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
// 3. Сохраняем слоты расписания
|
||||
if len(payload.Slots) > 0 {
|
||||
minStart, maxEnd := findSlotBounds(payload.Slots)
|
||||
if err := a.store.ReplaceOnecSlots(ctx, minStart, maxEnd, payload.Slots); err != nil {
|
||||
fail(w, http.StatusInternalServerError, "db_error", "Ошибка сохранения слотов: "+err.Error())
|
||||
if err := a.store.AcceptOnecSlots(ctx, from, to, payload.Slots); err != nil {
|
||||
handleErr(w, err)
|
||||
return
|
||||
}
|
||||
logID, err := a.store.StartOnecSync(ctx)
|
||||
if err == nil {
|
||||
_ = a.store.FinishOnecSync(ctx, logID, len(payload.Doctors), len(payload.Slots), "")
|
||||
}
|
||||
|
||||
ok(w, map[string]any{
|
||||
@ -127,7 +156,18 @@ func (a *API) onecPendingAppointments(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
commands, err := a.store.PendingOnecCommands(r.Context())
|
||||
if err != nil {
|
||||
handleErr(w, err)
|
||||
return
|
||||
}
|
||||
requested, err := a.store.OnecSnapshotRequested(r.Context())
|
||||
if err != nil {
|
||||
handleErr(w, err)
|
||||
return
|
||||
}
|
||||
ok(w, map[string]any{
|
||||
"commands": commands, "sync_requested": requested,
|
||||
"ok": true,
|
||||
"appointments": list,
|
||||
}, nil)
|
||||
@ -158,13 +198,17 @@ func (a *API) onecConfirmAppointment(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
if req.OnecAppointmentID == "" {
|
||||
if !uuidTextRE.MatchString(req.OnecAppointmentID) {
|
||||
fail(w, http.StatusBadRequest, "missing_param", "onec_appointment_id обязателен")
|
||||
return
|
||||
}
|
||||
|
||||
if err := a.store.ConfirmPendingAppointment(r.Context(), id, req.OnecAppointmentID, req.End); err != nil {
|
||||
fail(w, http.StatusInternalServerError, "db_error", err.Error())
|
||||
if errors.Is(err, storage.ErrOnecTransition) {
|
||||
fail(w, 409, "state_conflict", "Заявка уже обработана иначе")
|
||||
return
|
||||
}
|
||||
handleErr(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
@ -200,9 +244,66 @@ func (a *API) onecRejectAppointment(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
if err := a.store.RejectPendingAppointment(r.Context(), id, req.Reason); err != nil {
|
||||
if errors.Is(err, storage.ErrOnecTransition) {
|
||||
fail(w, 409, "state_conflict", "Заявка уже обработана иначе")
|
||||
return
|
||||
}
|
||||
fail(w, http.StatusInternalServerError, "db_error", err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
if a.bookingConfirmEnabled() {
|
||||
if err := a.store.MarkLeadOnly(r.Context(), id); err != nil {
|
||||
handleErr(w, err)
|
||||
return
|
||||
}
|
||||
a.kickConfirm()
|
||||
}
|
||||
ok(w, map[string]any{"ok": true, "id": id, "status": "failed"}, nil)
|
||||
}
|
||||
|
||||
func (a *API) onecCommandAck(w http.ResponseWriter, r *http.Request) {
|
||||
if !a.checkOnecKey(r) {
|
||||
fail(w, 401, "unauthorized", "Неверный ключ")
|
||||
return
|
||||
}
|
||||
id, err := strconv.ParseInt(r.PathValue("id"), 10, 64)
|
||||
if err != nil || id <= 0 {
|
||||
fail(w, 400, "bad_id", "Некорректная команда")
|
||||
return
|
||||
}
|
||||
var body struct {
|
||||
Success bool `json:"success"`
|
||||
Error string `json:"error"`
|
||||
}
|
||||
if err = json.NewDecoder(http.MaxBytesReader(w, r.Body, 8192)).Decode(&body); err != nil {
|
||||
fail(w, 400, "bad_json", "Некорректный JSON")
|
||||
return
|
||||
}
|
||||
if err = a.store.AckOnecCommand(r.Context(), id, body.Success, body.Error); err != nil {
|
||||
handleErr(w, err)
|
||||
return
|
||||
}
|
||||
ok(w, map[string]any{"ok": true}, nil)
|
||||
}
|
||||
|
||||
// Overlapping shifts can repeat the same key. A busy occurrence always wins.
|
||||
func mergeOnecSlots(slots []onec.Slot) []onec.Slot {
|
||||
out := make([]onec.Slot, 0, len(slots))
|
||||
seen := map[string]int{}
|
||||
for _, sl := range slots {
|
||||
key := sl.DoctorID + "/" + sl.BranchID + "/" + sl.Start.UTC().Format(time.RFC3339Nano)
|
||||
if i, ok := seen[key]; ok {
|
||||
if sl.Status == "busy" {
|
||||
out[i].Status = "busy"
|
||||
}
|
||||
if sl.End.After(out[i].End) {
|
||||
out[i].End = sl.End
|
||||
}
|
||||
continue
|
||||
}
|
||||
seen[key] = len(out)
|
||||
out = append(out, sl)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
@ -67,3 +67,17 @@ func TestOnecConfirmBadID(t *testing.T) {
|
||||
t.Fatalf("expected 400 bad id, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPushMergesDuplicateSlotsConservatively(t *testing.T) {
|
||||
start := time.Now().UTC()
|
||||
end := start.Add(time.Hour)
|
||||
free := onec.Slot{DoctorID: "d", BranchID: "b", Start: start, End: end, Status: "free"}
|
||||
busy := free
|
||||
busy.Status = "busy"
|
||||
for _, slots := range [][]onec.Slot{{free, busy}, {busy, free}} {
|
||||
got := mergeOnecSlots(slots)
|
||||
if len(got) != 1 || got[0].Status != "busy" {
|
||||
t.Fatalf("busy duplicate lost: %+v", got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -1,135 +1,81 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"crypto/subtle"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Входящие события от 1С. Опрос раз в 5 минут остаётся основным способом, вебхук только ускоряет обновление:
|
||||
// 1С сообщает, что расписание врача изменилось или запись сменила статус, а сайт сразу перечитывает данные из 1С.
|
||||
// Данные из тела события в расписание не пишутся — источник правды всегда ответ API 1С.
|
||||
|
||||
const maxOnecEventBody = 8 << 10
|
||||
|
||||
// WithOneCWebhook включает приём событий от 1С; пустой ключ — вебхук выключен.
|
||||
func (a *API) WithOneCWebhook(key string) { a.onecHook = strings.TrimSpace(key) }
|
||||
|
||||
var onecEventSyncBusy atomic.Bool
|
||||
func (a *API) onecAutonomousEnabled() bool { return a.onecHook != "" }
|
||||
|
||||
type onecEvent struct {
|
||||
Event string `json:"event"`
|
||||
DoctorID string `json:"doctor_id"`
|
||||
AppointmentID string `json:"appointment_id"`
|
||||
Status string `json:"status"`
|
||||
BranchID string `json:"branch_id"`
|
||||
Start *time.Time `json:"start"`
|
||||
End *time.Time `json:"end"`
|
||||
}
|
||||
|
||||
func sameKey(got, want string) bool {
|
||||
g, w := sha256.Sum256([]byte(got)), sha256.Sum256([]byte(want))
|
||||
return subtle.ConstantTimeCompare(g[:], w[:]) == 1
|
||||
}
|
||||
|
||||
func (a *API) onecEvents(w http.ResponseWriter, r *http.Request) {
|
||||
if a.onecHook == "" || !a.onec.Configured() {
|
||||
fail(w, http.StatusNotFound, "not_found", "Не найдено")
|
||||
if !a.onecAutonomousEnabled() {
|
||||
fail(w, 404, "not_found", "Не найдено")
|
||||
return
|
||||
}
|
||||
ip := clientIP(r)
|
||||
if a.tooMany(w, "onec-ev:"+ip, 300, time.Minute) {
|
||||
return
|
||||
}
|
||||
key := r.Header.Get("X-Api-Key")
|
||||
if key == "" {
|
||||
key = strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
|
||||
}
|
||||
if !sameKey(key, a.onecHook) {
|
||||
// неверные ключи считаем отдельно: перебор упирается в лимит
|
||||
if !a.checkOnecKey(r) {
|
||||
if a.tooMany(w, "onec-bad:"+ip, 20, 10*time.Minute) {
|
||||
return
|
||||
}
|
||||
fail(w, http.StatusUnauthorized, "unauthorized", "Неверный ключ")
|
||||
fail(w, 401, "unauthorized", "Неверный ключ")
|
||||
return
|
||||
}
|
||||
var ev onecEvent
|
||||
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxOnecEventBody)).Decode(&ev); err != nil {
|
||||
fail(w, http.StatusBadRequest, "bad_json", "Не удалось разобрать событие")
|
||||
fail(w, 400, "bad_json", "Не удалось разобрать событие")
|
||||
return
|
||||
}
|
||||
ev.DoctorID = strings.TrimSpace(ev.DoctorID)
|
||||
|
||||
switch ev.Event {
|
||||
case "schedule_changed":
|
||||
if ev.DoctorID == "" {
|
||||
a.onecFullResyncAsync(false)
|
||||
writeJSON(w, http.StatusAccepted, envelope{Data: map[string]any{"ok": true, "queued": "schedule"}})
|
||||
case "schedule_changed", "directory_changed":
|
||||
if err := a.store.RequestOnecSnapshot(r.Context()); err != nil {
|
||||
handleErr(w, err)
|
||||
return
|
||||
}
|
||||
if err := a.resyncOnecDoctor(r.Context(), ev.DoctorID); err != nil {
|
||||
fail(w, http.StatusBadGateway, "onec_sync", onecErrText(err))
|
||||
return
|
||||
}
|
||||
ok(w, map[string]any{"ok": true}, nil)
|
||||
|
||||
case "directory_changed":
|
||||
a.onecFullResyncAsync(true)
|
||||
writeJSON(w, http.StatusAccepted, envelope{Data: map[string]any{"ok": true, "queued": "directory"}})
|
||||
|
||||
writeJSON(w, http.StatusAccepted, envelope{Data: map[string]any{"ok": true, "queued": "push"}})
|
||||
case "appointment_status":
|
||||
if ev.AppointmentID == "" || (ev.Status != "booked" && ev.Status != "arrived" && ev.Status != "cancelled") {
|
||||
fail(w, http.StatusBadRequest, "validation", "Нужны appointment_id и status: booked, arrived или cancelled")
|
||||
fail(w, 400, "validation", "Нужны appointment_id и status: booked, arrived или cancelled")
|
||||
return
|
||||
}
|
||||
doctorID, _, found, err := a.store.SetOnecAppointmentStatus(r.Context(), ev.AppointmentID, ev.Status)
|
||||
if (ev.Start == nil) != (ev.End == nil) || (ev.Start != nil && !ev.Start.Before(*ev.End)) {
|
||||
fail(w, 400, "validation", "Некорректный интервал события")
|
||||
return
|
||||
}
|
||||
_, _, known, err := a.store.ApplyOnecAppointmentEvent(r.Context(), ev.AppointmentID, ev.Status, ev.DoctorID, ev.BranchID, ev.Start, ev.End)
|
||||
if err != nil {
|
||||
handleErr(w, err)
|
||||
return
|
||||
}
|
||||
if !found {
|
||||
// запись сделана не с сайта — нам она не нужна, но для 1С это не ошибка
|
||||
ok(w, map[string]any{"ok": true, "known": false}, nil)
|
||||
if err = a.store.RequestOnecSnapshot(r.Context()); err != nil {
|
||||
handleErr(w, err)
|
||||
return
|
||||
}
|
||||
if ev.Status == "cancelled" {
|
||||
if err := a.resyncOnecDoctor(r.Context(), doctorID); err != nil {
|
||||
slog.Warn("onec webhook: окна после отмены не обновились", "doctor", doctorID, "err", err)
|
||||
}
|
||||
}
|
||||
ok(w, map[string]any{"ok": true, "known": true}, nil)
|
||||
|
||||
ok(w, map[string]any{"ok": true, "known": known}, nil)
|
||||
default:
|
||||
fail(w, http.StatusBadRequest, "validation", "Неизвестное событие: допустимы schedule_changed, directory_changed, appointment_status")
|
||||
fail(w, 400, "validation", "Неизвестное событие: допустимы schedule_changed, directory_changed, appointment_status")
|
||||
}
|
||||
}
|
||||
|
||||
// resyncOnecDoctor заново забирает окна одного врача на весь горизонт.
|
||||
func (a *API) resyncOnecDoctor(ctx context.Context, onecID string) error {
|
||||
ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
defer cancel()
|
||||
from, to := onecWindow()
|
||||
slots, err := a.onec.Schedule(ctx, from, to, onecID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return a.store.ReplaceOnecDoctorSlots(ctx, onecID, from, to, slots)
|
||||
}
|
||||
|
||||
// onecFullResyncAsync запускает полную синхронизацию в фоне, чтобы 1С не ждала ответа; пока идёт одна — вторая не ставится.
|
||||
func (a *API) onecFullResyncAsync(dir bool) {
|
||||
if !onecEventSyncBusy.CompareAndSwap(false, true) {
|
||||
return
|
||||
}
|
||||
go func() {
|
||||
defer onecEventSyncBusy.Store(false)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
defer cancel()
|
||||
if err := a.syncOneC(ctx, dir); err != nil {
|
||||
slog.Warn("onec webhook sync", "err", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
@ -58,3 +58,11 @@ func TestOnecEventsBruteForceLimit(t *testing.T) {
|
||||
t.Fatalf("перебор ключа должен упираться в лимит, получили %d", last)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAutonomousWebhookValidationWithoutBridge(t *testing.T) {
|
||||
a := &API{limiter: newRateLimiter()}
|
||||
a.WithOneCWebhook("secret")
|
||||
if rec := postEvent(a, "secret", `{"event":"unsupported"}`); rec.Code != http.StatusBadRequest {
|
||||
t.Fatalf("bridge-free route disabled: %d", rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
23
internal/storage/migrations/0027_onec_autonomous.sql
Normal file
23
internal/storage/migrations/0027_onec_autonomous.sql
Normal file
@ -0,0 +1,23 @@
|
||||
-- Durable control plane for outbound-only 1C polling. No clinical data is removed.
|
||||
ALTER TABLE onec_appointments ADD COLUMN patient_comment text NOT NULL DEFAULT '';
|
||||
ALTER TABLE onec_appointments ADD COLUMN patient_birth_date text NOT NULL DEFAULT '';
|
||||
CREATE INDEX onec_appointments_occupancy_idx ON onec_appointments(doctor_onec_id,start_at)
|
||||
WHERE status IN ('pending','booked','arrived');
|
||||
CREATE TABLE onec_push_state (
|
||||
singleton boolean PRIMARY KEY DEFAULT true CHECK(singleton),
|
||||
last_snapshot_at timestamptz,
|
||||
sync_requested boolean NOT NULL DEFAULT true
|
||||
);
|
||||
INSERT INTO onec_push_state(singleton) VALUES(true);
|
||||
CREATE TABLE onec_commands (
|
||||
id bigserial PRIMARY KEY,
|
||||
appointment_id bigint NOT NULL REFERENCES onec_appointments(id),
|
||||
kind text NOT NULL CHECK(kind IN ('cancel','link_bitrix','link_crm')),
|
||||
value text NOT NULL DEFAULT '',
|
||||
final_state text NOT NULL DEFAULT '',
|
||||
completed_at timestamptz,
|
||||
notification_done boolean NOT NULL DEFAULT false,
|
||||
last_error text NOT NULL DEFAULT '',
|
||||
created_at timestamptz NOT NULL DEFAULT now(),
|
||||
UNIQUE(appointment_id,kind)
|
||||
);
|
||||
@ -2,6 +2,7 @@ package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@ -10,6 +11,9 @@ import (
|
||||
"dana/internal/onec"
|
||||
)
|
||||
|
||||
var ErrOnecSlotBusy = errors.New("slot already reserved")
|
||||
var ErrOnecTransition = errors.New("appointment state conflict")
|
||||
|
||||
var moscow = time.FixedZone("Europe/Moscow", 3*60*60)
|
||||
|
||||
// ---------- Справочники ----------
|
||||
@ -252,7 +256,10 @@ func (s *Storage) FreeOnecSlots(ctx context.Context, doctorOnecID string, from,
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT sl.start_at, sl.end_at, sl.branch_onec_id, COALESCE(NULLIF(b.address, ''), b.name, ''), sl.cabinet, sl.slot_id
|
||||
FROM onec_slots sl LEFT JOIN onec_branches b ON b.onec_id = sl.branch_onec_id
|
||||
WHERE sl.doctor_onec_id=$1 AND sl.status='free' AND sl.start_at >= greatest($2::timestamptz, now()) AND sl.start_at < $3
|
||||
WHERE sl.doctor_onec_id=$1 AND sl.status='free' AND sl.start_at >= greatest($2::timestamptz, now()) AND sl.start_at < $3 AND NOT EXISTS (
|
||||
SELECT 1 FROM onec_appointments a WHERE a.doctor_onec_id=sl.doctor_onec_id
|
||||
AND a.status IN ('pending','booked','arrived') AND a.start_at < sl.end_at
|
||||
AND COALESCE(a.end_at,a.start_at+interval '30 minutes') > sl.start_at)
|
||||
ORDER BY sl.start_at, sl.branch_onec_id`, doctorOnecID, from, to)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@ -356,8 +363,7 @@ func (s *Storage) OnecSyncStatus(ctx context.Context) (OnecSyncStatus, error) {
|
||||
// OnecFresh сообщает, была ли успешная синхронизация не раньше чем maxAge назад.
|
||||
func (s *Storage) OnecFresh(ctx context.Context, maxAge time.Duration) bool {
|
||||
var ok bool
|
||||
_ = s.pool.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM onec_sync_log WHERE finished_at > now() - $1::interval AND error = '')`,
|
||||
maxAge.String()).Scan(&ok)
|
||||
_ = s.pool.QueryRow(ctx, `SELECT last_snapshot_at > now() - $1::interval FROM onec_push_state WHERE singleton`, maxAge.String()).Scan(&ok)
|
||||
return ok
|
||||
}
|
||||
|
||||
@ -507,6 +513,8 @@ type OnecAppointment struct {
|
||||
ServiceName string `json:"service_name"`
|
||||
ServiceID string `json:"-"`
|
||||
BranchID string `json:"-"`
|
||||
Comment string `json:"-"`
|
||||
BirthDate string `json:"-"`
|
||||
// подтверждение оператором в Битрикс24 (см. booking_confirm.go)
|
||||
ConfirmState string `json:"confirm_state"`
|
||||
BitrixLeadID string `json:"bitrix_lead_id"`
|
||||
@ -520,20 +528,58 @@ type OnecAppointment struct {
|
||||
// BeginOnecAppointment создаёт запись в статусе pending. Если такой request_id уже был, возвращает прежнюю запись
|
||||
// и created=false — так повтор запроса не создаёт вторую запись в 1С.
|
||||
func (s *Storage) BeginOnecAppointment(ctx context.Context, a OnecAppointment) (rec OnecAppointment, created bool, err error) {
|
||||
err = s.pool.QueryRow(ctx, `
|
||||
INSERT INTO onec_appointments (request_id, doctor_onec_id, branch_onec_id, service_onec_id, start_at, patient_name, patient_phone)
|
||||
VALUES ($1,$2,$3,$4,$5,$6,$7) ON CONFLICT (request_id) DO NOTHING RETURNING id`,
|
||||
a.RequestID, a.DoctorOnec, a.BranchID, a.ServiceID, a.Start, a.Patient, a.Phone).Scan(&a.ID)
|
||||
if err == nil {
|
||||
a.Status = "pending"
|
||||
return a, true, nil
|
||||
}
|
||||
if wrapNotFound(err) != ErrNotFound {
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return a, false, err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
// Request lock makes retries safe even if doctor/slot in the request changed.
|
||||
if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, "onec-request:"+a.RequestID); err != nil {
|
||||
return a, false, err
|
||||
}
|
||||
var existing bool
|
||||
if err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM onec_appointments WHERE request_id=$1)`, a.RequestID).Scan(&existing); err != nil {
|
||||
return a, false, err
|
||||
}
|
||||
if existing {
|
||||
rec, err = s.onecAppointmentByRequest(ctx, a.RequestID)
|
||||
return rec, false, err
|
||||
}
|
||||
if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, "onec-doctor:"+a.DoctorOnec); err != nil {
|
||||
return a, false, err
|
||||
}
|
||||
var end time.Time
|
||||
if a.Status != "failed" {
|
||||
a.Status = "pending"
|
||||
}
|
||||
if a.Status == "pending" {
|
||||
err = tx.QueryRow(ctx, `SELECT sl.branch_onec_id,sl.end_at FROM onec_slots sl
|
||||
WHERE sl.doctor_onec_id=$1 AND sl.start_at=$2 AND ($3='' OR sl.branch_onec_id=$3)
|
||||
AND sl.status='free' AND NOT EXISTS (
|
||||
SELECT 1 FROM onec_appointments a WHERE a.doctor_onec_id=sl.doctor_onec_id
|
||||
AND a.status IN ('pending','booked','arrived') AND a.start_at < sl.end_at
|
||||
AND COALESCE(a.end_at,a.start_at+interval '30 minutes') > sl.start_at)
|
||||
ORDER BY sl.branch_onec_id LIMIT 1`, a.DoctorOnec, a.Start, a.BranchID).Scan(&a.BranchID, &end)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return a, false, ErrOnecSlotBusy
|
||||
}
|
||||
if err != nil {
|
||||
return a, false, err
|
||||
}
|
||||
a.End = &end
|
||||
}
|
||||
err = tx.QueryRow(ctx, `INSERT INTO onec_appointments
|
||||
(request_id,doctor_onec_id,branch_onec_id,service_onec_id,start_at,end_at,patient_name,patient_phone,patient_comment,patient_birth_date,status)
|
||||
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) RETURNING id`,
|
||||
a.RequestID, a.DoctorOnec, a.BranchID, a.ServiceID, a.Start, a.End, a.Patient, a.Phone, a.Comment, a.BirthDate, a.Status).Scan(&a.ID)
|
||||
if err != nil {
|
||||
return a, false, err
|
||||
}
|
||||
if err = tx.Commit(ctx); err != nil {
|
||||
return a, false, err
|
||||
}
|
||||
return a, true, nil
|
||||
}
|
||||
|
||||
func (s *Storage) onecAppointmentByRequest(ctx context.Context, requestID string) (OnecAppointment, error) {
|
||||
var r OnecAppointment
|
||||
@ -585,10 +631,16 @@ func (s *Storage) OnecAppointments(ctx context.Context, limit int) ([]OnecAppoin
|
||||
// SetOnecAppointmentStatus обновляет статус записи по номеру записи в 1С (вебхук 1С).
|
||||
// Возвращает врача и время приёма, чтобы пересчитать его окна.
|
||||
func (s *Storage) SetOnecAppointmentStatus(ctx context.Context, onecAppointmentID, status string) (doctorOnecID string, start time.Time, found bool, err error) {
|
||||
return s.ApplyOnecAppointmentEvent(ctx, onecAppointmentID, status, "", "", nil, nil)
|
||||
}
|
||||
func (s *Storage) ApplyOnecAppointmentEvent(ctx context.Context, onecAppointmentID, status, doctor, branch string, eventStart, eventEnd *time.Time) (doctorOnecID string, start time.Time, found bool, err error) {
|
||||
err = s.pool.QueryRow(ctx, `
|
||||
UPDATE onec_appointments SET status=$2, updated_at=now()
|
||||
WHERE onec_appointment_id=$1 AND status IN ('booked','arrived','cancelled')
|
||||
RETURNING doctor_onec_id, start_at`, onecAppointmentID, status).Scan(&doctorOnecID, &start)
|
||||
UPDATE onec_appointments SET status=$2, updated_at=now(),
|
||||
doctor_onec_id=COALESCE(NULLIF($3,''),doctor_onec_id),branch_onec_id=COALESCE(NULLIF($4,''),branch_onec_id),
|
||||
start_at=COALESCE($5,start_at),end_at=COALESCE($6,end_at),
|
||||
confirm_state=CASE WHEN $2='cancelled' AND confirm_state<>'expired' THEN 'cancelled' ELSE confirm_state END
|
||||
WHERE onec_appointment_id=$1 AND status IN ('booked','arrived','cancelled') AND (status<>'cancelled' OR $2='cancelled') AND (status<>'arrived' OR $2<>'booked')
|
||||
RETURNING doctor_onec_id, start_at`, onecAppointmentID, status, doctor, branch, eventStart, eventEnd).Scan(&doctorOnecID, &start)
|
||||
if err != nil {
|
||||
if wrapNotFound(err) == ErrNotFound {
|
||||
return "", time.Time{}, false, nil
|
||||
@ -608,6 +660,8 @@ type OnecPendingAppointment struct {
|
||||
Patient string `json:"patient_name"`
|
||||
Phone string `json:"patient_phone"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
Comment string `json:"comment"`
|
||||
BirthDate string `json:"birth_date"`
|
||||
}
|
||||
|
||||
func (s *Storage) PendingOnecAppointments(ctx context.Context, limit int) ([]OnecPendingAppointment, error) {
|
||||
@ -615,7 +669,7 @@ func (s *Storage) PendingOnecAppointments(ctx context.Context, limit int) ([]One
|
||||
limit = 20
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT id, request_id, doctor_onec_id, branch_onec_id, COALESCE(service_onec_id, ''), start_at, patient_name, patient_phone, created_at
|
||||
SELECT id, request_id, doctor_onec_id, branch_onec_id, COALESCE(service_onec_id, ''), start_at, patient_name, patient_phone, created_at, patient_comment, patient_birth_date
|
||||
FROM onec_appointments
|
||||
WHERE status = 'pending'
|
||||
ORDER BY id ASC
|
||||
@ -627,26 +681,39 @@ func (s *Storage) PendingOnecAppointments(ctx context.Context, limit int) ([]One
|
||||
out := []OnecPendingAppointment{}
|
||||
for rows.Next() {
|
||||
var a OnecPendingAppointment
|
||||
if err := rows.Scan(&a.ID, &a.RequestID, &a.DoctorID, &a.BranchID, &a.ServiceID, &a.Start, &a.Patient, &a.Phone, &a.CreatedAt); err != nil {
|
||||
if err := rows.Scan(&a.ID, &a.RequestID, &a.DoctorID, &a.BranchID, &a.ServiceID, &a.Start, &a.Patient, &a.Phone, &a.CreatedAt, &a.Comment, &a.BirthDate); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
a.Start = a.Start.In(moscow)
|
||||
out = append(out, a)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (s *Storage) ConfirmPendingAppointment(ctx context.Context, id int64, onecID string, end *time.Time) error {
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
UPDATE onec_appointments
|
||||
SET status = 'booked', onec_appointment_id = $2, end_at = $3, updated_at = now()
|
||||
WHERE id = $1 AND status = 'pending'`, id, onecID, end)
|
||||
tag, err := s.pool.Exec(ctx, `UPDATE onec_appointments SET status='booked',onec_appointment_id=$2,
|
||||
end_at=COALESCE($3,end_at),updated_at=now() WHERE id=$1 AND
|
||||
COALESCE($3,end_at)>start_at AND (status='pending' OR (status='booked' AND onec_appointment_id=$2 AND ($3::timestamptz IS NULL OR end_at=$3)))`, id, onecID, end)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if tag.RowsAffected() == 0 {
|
||||
return ErrOnecTransition
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (s *Storage) RejectPendingAppointment(ctx context.Context, id int64, reason string) error {
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
UPDATE onec_appointments
|
||||
SET status = 'failed', error = $2, updated_at = now()
|
||||
WHERE id = $1 AND status = 'pending'`, id, reason)
|
||||
tag, err := s.pool.Exec(ctx, `UPDATE onec_appointments SET status='failed',error=$2,updated_at=now()
|
||||
WHERE id=$1 AND status IN ('pending','failed')`, id, reason)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return ErrOnecTransition
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (s *Storage) AttachOnecLead(ctx context.Context, id int64, lead *int64) error {
|
||||
_, err := s.pool.Exec(ctx, `UPDATE onec_appointments SET lead_id=COALESCE($2,lead_id) WHERE id=$1`, id, lead)
|
||||
return err
|
||||
}
|
||||
|
||||
157
internal/storage/onec_autonomous.go
Normal file
157
internal/storage/onec_autonomous.go
Normal file
@ -0,0 +1,157 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"dana/internal/onec"
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
func (s *Storage) RequestOnecSnapshot(ctx context.Context) error {
|
||||
_, err := s.pool.Exec(ctx, `UPDATE onec_push_state SET sync_requested=true WHERE singleton`)
|
||||
return err
|
||||
}
|
||||
func (s *Storage) OnecSnapshotRequested(ctx context.Context) (bool, error) {
|
||||
var v bool
|
||||
err := s.pool.QueryRow(ctx, `SELECT sync_requested FROM onec_push_state WHERE singleton`).Scan(&v)
|
||||
return v, err
|
||||
}
|
||||
|
||||
// Replacing even an empty horizon and marking it fresh are one transaction.
|
||||
func (s *Storage) AcceptOnecSlots(ctx context.Context, from, to time.Time, slots []onec.Slot) error {
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(719472109)`); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = tx.Exec(ctx, `DELETE FROM onec_slots WHERE (start_at >= $1 AND start_at < $2) OR start_at < now()-interval '1 day'`, from, to); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = insertSlots(ctx, tx, from, to, slots); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = tx.Exec(ctx, `UPDATE onec_push_state SET last_snapshot_at=now(),sync_requested=false WHERE singleton`); err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Commit(ctx)
|
||||
}
|
||||
|
||||
type OnecCommand struct {
|
||||
ID int64 `json:"id"`
|
||||
Kind string `json:"kind"`
|
||||
AppointmentID string `json:"appointment_id"`
|
||||
Value string `json:"value"`
|
||||
}
|
||||
|
||||
func (s *Storage) QueueOnecCommand(ctx context.Context, appointmentID int64, kind, value, finalState string) error {
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
_, err = tx.Exec(ctx, `INSERT INTO onec_commands(appointment_id,kind,value,final_state)
|
||||
SELECT id,$2,$3,$4 FROM onec_appointments WHERE id=$1 AND onec_appointment_id<>''
|
||||
ON CONFLICT(appointment_id,kind) DO UPDATE SET value=EXCLUDED.value
|
||||
WHERE onec_commands.completed_at IS NULL`, appointmentID, kind, value, finalState)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if kind == "cancel" {
|
||||
if _, err = tx.Exec(ctx, `UPDATE onec_appointments SET confirm_state='cancel_pending',updated_at=now() WHERE id=$1 AND status='booked'
|
||||
AND EXISTS(SELECT 1 FROM onec_commands WHERE appointment_id=$1 AND kind='cancel' AND completed_at IS NULL)`, appointmentID); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return tx.Commit(ctx)
|
||||
}
|
||||
|
||||
func (s *Storage) PendingOnecCommands(ctx context.Context) ([]OnecCommand, error) {
|
||||
rows, err := s.pool.Query(ctx, `SELECT c.id,c.kind,a.onec_appointment_id,c.value FROM onec_commands c
|
||||
JOIN onec_appointments a ON a.id=c.appointment_id WHERE c.completed_at IS NULL ORDER BY c.id LIMIT 50`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := []OnecCommand{}
|
||||
for rows.Next() {
|
||||
var c OnecCommand
|
||||
if err = rows.Scan(&c.ID, &c.Kind, &c.AppointmentID, &c.Value); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, c)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
func (s *Storage) AckOnecCommand(ctx context.Context, id int64, success bool, message string) error {
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
var kind, state string
|
||||
var appointmentID int64
|
||||
var completed *time.Time
|
||||
err = tx.QueryRow(ctx, `SELECT kind,appointment_id,final_state,completed_at FROM onec_commands WHERE id=$1 FOR UPDATE`, id).Scan(&kind, &appointmentID, &state, &completed)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return ErrNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if completed != nil {
|
||||
return nil
|
||||
}
|
||||
if !success {
|
||||
_, err = tx.Exec(ctx, `UPDATE onec_commands SET last_error=$2 WHERE id=$1`, id, message)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Commit(ctx)
|
||||
}
|
||||
if kind == "cancel" {
|
||||
_, err = tx.Exec(ctx, `UPDATE onec_appointments SET status='cancelled',confirm_state=$2,confirm_error='',updated_at=now() WHERE id=$1 AND status IN ('booked','cancelled')`, appointmentID, state)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = tx.Exec(ctx, `UPDATE onec_push_state SET sync_requested=true WHERE singleton`); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if _, err = tx.Exec(ctx, `UPDATE onec_commands SET completed_at=now(),last_error='' WHERE id=$1`, id); err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Commit(ctx)
|
||||
}
|
||||
|
||||
// Expiry in B24 is retried independently after 1C has confirmed the cancellation.
|
||||
type ExpiredOnecLead struct {
|
||||
CommandID int64
|
||||
LeadID string
|
||||
}
|
||||
|
||||
func (s *Storage) ExpiredOnecLeads(ctx context.Context) ([]ExpiredOnecLead, error) {
|
||||
rows, err := s.pool.Query(ctx, `SELECT c.id,a.bitrix_lead_id FROM onec_commands c JOIN onec_appointments a ON a.id=c.appointment_id
|
||||
WHERE c.kind='cancel' AND c.final_state='expired' AND c.completed_at IS NOT NULL AND NOT c.notification_done AND a.bitrix_lead_id<>'' LIMIT 100`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := []ExpiredOnecLead{}
|
||||
for rows.Next() {
|
||||
var v ExpiredOnecLead
|
||||
if err = rows.Scan(&v.CommandID, &v.LeadID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, v)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
func (s *Storage) MarkExpiredOnecLead(ctx context.Context, id int64) error {
|
||||
_, err := s.pool.Exec(ctx, `UPDATE onec_commands SET notification_done=true WHERE id=$1`, id)
|
||||
return err
|
||||
}
|
||||
149
internal/storage/onec_autonomous_test.go
Normal file
149
internal/storage/onec_autonomous_test.go
Normal file
@ -0,0 +1,149 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"dana/internal/onec"
|
||||
)
|
||||
|
||||
// Run only against the disposable DB created by the autonomy acceptance command.
|
||||
func TestAutonomousBookingLifecycle(t *testing.T) {
|
||||
dsn := os.Getenv("DANA_AUTONOMY_TEST_DSN")
|
||||
if dsn == "" {
|
||||
t.Skip("requires disposable autonomy database")
|
||||
}
|
||||
ctx := context.Background()
|
||||
s, err := New(ctx, dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer s.Close()
|
||||
if err = s.Migrate(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
must := func(err error) {
|
||||
t.Helper()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
doctor := fmt.Sprintf("autonomy-%d", time.Now().UnixNano())
|
||||
start := time.Now().UTC().Truncate(time.Hour).Add(48 * time.Hour)
|
||||
end := start.Add(30 * time.Minute)
|
||||
slot := onec.Slot{DoctorID: doctor, BranchID: "branch", Start: start, End: end, Status: "free"}
|
||||
from, to := start.Add(-time.Hour), end.Add(time.Hour)
|
||||
must(s.AcceptOnecSlots(ctx, from, to, []onec.Slot{slot}))
|
||||
if !s.OnecFresh(ctx, time.Hour) {
|
||||
t.Fatal("push did not mark freshness")
|
||||
}
|
||||
var wins atomic.Int32
|
||||
var winner OnecAppointment
|
||||
var mu sync.Mutex
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 12; i++ {
|
||||
wg.Add(1)
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
a, created, err := s.BeginOnecAppointment(ctx, OnecAppointment{RequestID: fmt.Sprintf("%s-%d", doctor, i), DoctorOnec: doctor, BranchID: "branch", Start: start, Patient: "test", Phone: "test", Comment: "test no calls"})
|
||||
if errors.Is(err, ErrOnecSlotBusy) {
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
return
|
||||
}
|
||||
if !created {
|
||||
t.Error("unexpected duplicate")
|
||||
return
|
||||
}
|
||||
wins.Add(1)
|
||||
mu.Lock()
|
||||
winner = a
|
||||
mu.Unlock()
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
if wins.Load() != 1 {
|
||||
t.Fatalf("concurrent reservation winners=%d", wins.Load())
|
||||
}
|
||||
assertFree := func(n int) {
|
||||
t.Helper()
|
||||
v, err := s.FreeOnecSlots(ctx, doctor, from, to)
|
||||
must(err)
|
||||
if len(v) != n {
|
||||
t.Fatalf("free slots=%d want %d", len(v), n)
|
||||
}
|
||||
}
|
||||
assertFree(0)
|
||||
same, created, err := s.BeginOnecAppointment(ctx, OnecAppointment{RequestID: winner.RequestID, DoctorOnec: doctor, Start: start})
|
||||
must(err)
|
||||
if created || same.ID != winner.ID {
|
||||
t.Fatal("request retry duplicated booking")
|
||||
}
|
||||
must(s.AcceptOnecSlots(ctx, from, to, []onec.Slot{slot}))
|
||||
assertFree(0) // stale free snapshot cannot reopen pending
|
||||
pending, err := s.PendingOnecAppointments(ctx, 50)
|
||||
must(err)
|
||||
found := false
|
||||
for _, p := range pending {
|
||||
if p.ID == winner.ID {
|
||||
found = true
|
||||
if p.Start.Format("-07:00") != "+03:00" || p.Comment != "test no calls" {
|
||||
t.Fatalf("bad pull fields: %+v", p)
|
||||
}
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatal("pending booking missing")
|
||||
}
|
||||
uuid := "00000000-0000-0000-0000-000000000001"
|
||||
must(s.ConfirmPendingAppointment(ctx, winner.ID, uuid, &end))
|
||||
must(s.ConfirmPendingAppointment(ctx, winner.ID, uuid, &end))
|
||||
if !errors.Is(s.ConfirmPendingAppointment(ctx, winner.ID, "different", &end), ErrOnecTransition) {
|
||||
t.Fatal("mismatched confirm accepted")
|
||||
}
|
||||
if !errors.Is(s.RejectPendingAppointment(ctx, winner.ID, "late reject"), ErrOnecTransition) {
|
||||
t.Fatal("late reject accepted")
|
||||
}
|
||||
must(s.AcceptOnecSlots(ctx, from, to, []onec.Slot{slot}))
|
||||
assertFree(0) // booked also survives full Push
|
||||
must(s.QueueOnecCommand(ctx, winner.ID, "link_bitrix", "123", ""))
|
||||
must(s.QueueOnecCommand(ctx, winner.ID, "link_bitrix", "123", ""))
|
||||
commands, err := s.PendingOnecCommands(ctx)
|
||||
must(err)
|
||||
if len(commands) != 1 {
|
||||
t.Fatal("duplicate link commands")
|
||||
}
|
||||
must(s.AckOnecCommand(ctx, commands[0].ID, false, "retry"))
|
||||
commands, err = s.PendingOnecCommands(ctx)
|
||||
must(err)
|
||||
if len(commands) != 1 {
|
||||
t.Fatal("failed ACK lost command")
|
||||
}
|
||||
must(s.AckOnecCommand(ctx, commands[0].ID, true, ""))
|
||||
must(s.AckOnecCommand(ctx, commands[0].ID, true, ""))
|
||||
must(s.QueueOnecCommand(ctx, winner.ID, "cancel", "", "cancelled"))
|
||||
assertFree(0)
|
||||
commands, err = s.PendingOnecCommands(ctx)
|
||||
must(err)
|
||||
if len(commands) != 1 || commands[0].Kind != "cancel" {
|
||||
t.Fatal("cancel not queued")
|
||||
}
|
||||
must(s.AckOnecCommand(ctx, commands[0].ID, true, ""))
|
||||
assertFree(1)
|
||||
_, _, known, err := s.SetOnecAppointmentStatus(ctx, uuid, "booked")
|
||||
must(err)
|
||||
if known {
|
||||
t.Fatal("delayed booked hook resurrected cancellation")
|
||||
}
|
||||
assertFree(1)
|
||||
must(s.AcceptOnecSlots(ctx, from, to, []onec.Slot{}))
|
||||
assertFree(0) // explicit empty horizon clears old shifts
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user