dana-medical-site/internal/storage/onec_autonomous.go

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
}