package at.procon.dip.domain.time.service;

import at.procon.dip.domain.access.DocumentVisibility;
import at.procon.dip.domain.document.DocumentFamily;
import at.procon.dip.domain.document.DocumentStatus;
import at.procon.dip.domain.document.DocumentType;
import at.procon.dip.domain.document.entity.Document;
import at.procon.dip.domain.document.repository.DocumentRepository;
import at.procon.dip.domain.time.config.TimeDomainProperties;
import at.procon.dip.domain.time.entity.*;
import at.procon.dip.domain.time.entity.leitstand.LeitstandPerson;
import at.procon.dip.domain.time.entity.leitstand.LeitstandTimeRecording;
import at.procon.dip.domain.time.leitstand.source.LeitstandSourceRows;
import at.procon.dip.domain.time.leitstand.source.LeitstandTimeSourceClient;
import at.procon.dip.domain.time.repository.TimeEntryRepository;
import at.procon.dip.domain.time.repository.TimeEntrySourceLinkRepository;
import at.procon.dip.domain.time.repository.TimeSyncRunRepository;
import at.procon.dip.domain.time.repository.TimeSyncStateRepository;
import at.procon.dip.domain.time.repository.leitstand.LeitstandPersonRepository;
import at.procon.dip.domain.time.repository.leitstand.LeitstandTimeRecordingRepository;
import at.procon.dip.runtime.condition.ConditionalOnRuntimeMode;
import at.procon.dip.runtime.config.RuntimeMode;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.PreparedStatement;
import java.sql.Timestamp;
import java.time.OffsetDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

@Service
//@ConditionalOnRuntimeMode(RuntimeMode.NEW)
@ConditionalOnProperty(prefix = "dip.time.leitstand", name = "enabled", havingValue = "true")
@RequiredArgsConstructor
@Slf4j
public class LeitstandTimeImportService {

    private static final String GLOBAL_SCOPE = "leitstand:all";

    private final TimeDomainProperties properties;
    private final LeitstandTimeSourceClient sourceClient;
    private final JdbcTemplate jdbcTemplate;
    private final TimeSyncRunRepository syncRunRepository;
    private final TimeSyncStateRepository syncStateRepository;
    private final DocumentRepository documentRepository;
    private final TimeEntryRepository timeEntryRepository;
    private final TimeEntrySourceLinkRepository sourceLinkRepository;
    private final LeitstandTimeRecordingRepository timeRecordingRepository;
    private final LeitstandPersonRepository personRepository;
    private final LeitstandTimeProjectionService projectionService;

    public void runSync() {
        TimeSyncRun run = syncRunRepository.save(TimeSyncRun.builder()
                .sourceSystem(TimeSourceSystem.LEITSTAND)
                .runType(syncStateRepository.findBySourceSystemAndScopeKey(TimeSourceSystem.LEITSTAND, GLOBAL_SCOPE).isPresent() ? TimeSyncRunType.INCREMENTAL : TimeSyncRunType.INITIAL)
                .scopeKey(properties.getLeitstand().getScopeKey())
                .status(TimeSyncRunStatus.RUNNING)
                .initiatedBy("startup-runner")
                .build());
        try {
            syncActivityTypes(run);
            syncPersons(run);
            syncOrganizations(run);
            syncContracts(run);
            syncContractPositions(run);
            syncCostUnits(run);
            syncTasks(run);
            syncPersonTaskAssignments(run);
            List<LeitstandSourceRows.TimeRecordingRow> recordings = syncTimeRecordings(run);
            syncTimeRecordingAssignments(run);
            if (properties.getLeitstand().isCreateCanonicalTimeEntries()) {
                upsertCanonicalTimeEntries(recordings);
                if (properties.getLeitstand().isBuildSearchProjection()) {
                    projectionService.refreshAll();
                }
            }
            run.setStatus(TimeSyncRunStatus.COMPLETED);
            run.setFinishedAt(OffsetDateTime.now());
            syncRunRepository.save(run);
            updateGlobalState(run, null);
        } catch (Exception ex) {
            run.setStatus(TimeSyncRunStatus.FAILED);
            run.setFinishedAt(OffsetDateTime.now());
            run.setErrorMessage(ex.getMessage());
            syncRunRepository.save(run);
            updateGlobalState(run, ex.getMessage());
            throw ex;
        }
    }

    private void syncActivityTypes(TimeSyncRun run) {
        List<LeitstandSourceRows.ActivityTypeRow> rows = sourceClient.fetchActivityTypes();
        run.setRowsRead(run.getRowsRead() + rows.size());
        jdbcTemplate.batchUpdate("""
                INSERT INTO TIME.ls_activity_type(id, l_code, bez)
                VALUES (?, ?, ?)
                ON CONFLICT (id) DO UPDATE SET l_code = EXCLUDED.l_code, bez = EXCLUDED.bez
                """, new BatchPreparedStatementSetter() {
            public void setValues(PreparedStatement ps, int i) throws java.sql.SQLException {
                var r = rows.get(i);
                ps.setObject(1, r.id()); ps.setString(2, r.lCode()); ps.setString(3, r.bez());
            }
            public int getBatchSize() { return rows.size(); }
        });
        updateTableState("activity-type", null, rows.isEmpty() ? null : String.valueOf(rows.get(rows.size()-1).id()));
    }

    private void syncPersons(TimeSyncRun run) {
        String watermark = watermark("person");
        List<LeitstandSourceRows.PersonRow> rows = sourceClient.fetchPersons(watermark);
        run.setRowsRead(run.getRowsRead() + rows.size());
        batchUpsert("""
                INSERT INTO TIME.ls_person(dbk, dbk_uid, upd, upd_anz, upd_uid, status, pers_nr, last_name, first_name, category_dbk, salutation, title, street, postal_code, city, country_code, phone, fax, email, cost_per_hour, organization_dbk)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT (dbk) DO UPDATE SET dbk_uid=EXCLUDED.dbk_uid, upd=EXCLUDED.upd, upd_anz=EXCLUDED.upd_anz, upd_uid=EXCLUDED.upd_uid, status=EXCLUDED.status, pers_nr=EXCLUDED.pers_nr, last_name=EXCLUDED.last_name, first_name=EXCLUDED.first_name, category_dbk=EXCLUDED.category_dbk, salutation=EXCLUDED.salutation, title=EXCLUDED.title, street=EXCLUDED.street, postal_code=EXCLUDED.postal_code, city=EXCLUDED.city, country_code=EXCLUDED.country_code, phone=EXCLUDED.phone, fax=EXCLUDED.fax, email=EXCLUDED.email, cost_per_hour=EXCLUDED.cost_per_hour, organization_dbk=EXCLUDED.organization_dbk, updated_at=CURRENT_TIMESTAMP
                """, rows, (ps, r) -> { set(ps,1,r.dbk()); set(ps,2,r.dbkUid()); set(ps,3,r.upd()); set(ps,4,r.updAnz()); set(ps,5,r.updUid()); set(ps,6,r.status()); set(ps,7,r.personNumber()); set(ps,8,r.lastName()); set(ps,9,r.firstName()); set(ps,10,r.categoryDbk()); set(ps,11,r.salutation()); set(ps,12,r.title()); set(ps,13,r.street()); set(ps,14,r.postalCode()); set(ps,15,r.city()); set(ps,16,r.countryCode()); set(ps,17,r.phone()); set(ps,18,r.fax()); set(ps,19,r.email()); set(ps,20,r.costPerHour()); set(ps,21,r.organizationDbk());});
        updateTableState("person", maxUpd(rows, LeitstandSourceRows.PersonRow::upd), lastId(rows, LeitstandSourceRows.PersonRow::dbk));
    }

    private void syncOrganizations(TimeSyncRun run) { var rows = sourceClient.fetchOrganizations(watermark("organization")); run.setRowsRead(run.getRowsRead()+rows.size()); batchUpsert("INSERT INTO TIME.ls_organization(dbk, dbk_uid, upd, upd_anz, upd_uid, status, betreu, name, street, postal_code, city, phone, fax, email, org_nr, short_name) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (dbk) DO UPDATE SET dbk_uid=EXCLUDED.dbk_uid, upd=EXCLUDED.upd, upd_anz=EXCLUDED.upd_anz, upd_uid=EXCLUDED.upd_uid, status=EXCLUDED.status, betreu=EXCLUDED.betreu, name=EXCLUDED.name, street=EXCLUDED.street, postal_code=EXCLUDED.postal_code, city=EXCLUDED.city, phone=EXCLUDED.phone, fax=EXCLUDED.fax, email=EXCLUDED.email, org_nr=EXCLUDED.org_nr, short_name=EXCLUDED.short_name, updated_at=CURRENT_TIMESTAMP", rows, (ps,r)->{set(ps,1,r.dbk());set(ps,2,r.dbkUid());set(ps,3,r.upd());set(ps,4,r.updAnz());set(ps,5,r.updUid());set(ps,6,r.status());set(ps,7,r.betreu());set(ps,8,r.name());set(ps,9,r.street());set(ps,10,r.postalCode());set(ps,11,r.city());set(ps,12,r.phone());set(ps,13,r.fax());set(ps,14,r.email());set(ps,15,r.orgNumber());set(ps,16,r.shortName());}); updateTableState("organization", maxUpd(rows, LeitstandSourceRows.OrganizationRow::upd), lastId(rows, LeitstandSourceRows.OrganizationRow::dbk)); }
    private void syncContracts(TimeSyncRun run) { var rows = sourceClient.fetchContracts(watermark("contract")); run.setRowsRead(run.getRowsRead()+rows.size()); batchUpsert("INSERT INTO TIME.ls_contract(dbk, dbk_uid, upd, upd_anz, upd_uid, status, organization_dbk, lfdnr, name, iref, eref, description, valid_from, valid_to) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (dbk) DO UPDATE SET dbk_uid=EXCLUDED.dbk_uid, upd=EXCLUDED.upd, upd_anz=EXCLUDED.upd_anz, upd_uid=EXCLUDED.upd_uid, status=EXCLUDED.status, organization_dbk=EXCLUDED.organization_dbk, lfdnr=EXCLUDED.lfdnr, name=EXCLUDED.name, iref=EXCLUDED.iref, eref=EXCLUDED.eref, description=EXCLUDED.description, valid_from=EXCLUDED.valid_from, valid_to=EXCLUDED.valid_to, updated_at=CURRENT_TIMESTAMP", rows, (ps,r)->{set(ps,1,r.dbk());set(ps,2,r.dbkUid());set(ps,3,r.upd());set(ps,4,r.updAnz());set(ps,5,r.updUid());set(ps,6,r.status());set(ps,7,r.organizationDbk());set(ps,8,r.lfdnr());set(ps,9,r.name());set(ps,10,r.iref());set(ps,11,r.eref());set(ps,12,r.description());set(ps,13,ts(r.validFrom()));set(ps,14,ts(r.validTo()));}); updateTableState("contract", maxUpd(rows, LeitstandSourceRows.ContractRow::upd), lastId(rows, LeitstandSourceRows.ContractRow::dbk)); }
    private void syncContractPositions(TimeSyncRun run) { var rows = sourceClient.fetchContractPositions(watermark("contract-position")); run.setRowsRead(run.getRowsRead()+rows.size()); batchUpsert("INSERT INTO TIME.ls_contract_position(dbk, dbk_uid, upd, upd_anz, upd_uid, status, contract_dbk, lfdnr, project_name, sales_price, purchase_price, description, valid_from, valid_to, name, iref, eref) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (dbk) DO UPDATE SET dbk_uid=EXCLUDED.dbk_uid, upd=EXCLUDED.upd, upd_anz=EXCLUDED.upd_anz, upd_uid=EXCLUDED.upd_uid, status=EXCLUDED.status, contract_dbk=EXCLUDED.contract_dbk, lfdnr=EXCLUDED.lfdnr, project_name=EXCLUDED.project_name, sales_price=EXCLUDED.sales_price, purchase_price=EXCLUDED.purchase_price, description=EXCLUDED.description, valid_from=EXCLUDED.valid_from, valid_to=EXCLUDED.valid_to, name=EXCLUDED.name, iref=EXCLUDED.iref, eref=EXCLUDED.eref, updated_at=CURRENT_TIMESTAMP", rows, (ps,r)->{set(ps,1,r.dbk());set(ps,2,r.dbkUid());set(ps,3,r.upd());set(ps,4,r.updAnz());set(ps,5,r.updUid());set(ps,6,r.status());set(ps,7,r.contractDbk());set(ps,8,r.lfdnr());set(ps,9,r.projectName());set(ps,10,r.salesPrice());set(ps,11,r.purchasePrice());set(ps,12,r.description());set(ps,13,ts(r.validFrom()));set(ps,14,ts(r.validTo()));set(ps,15,r.name());set(ps,16,r.iref());set(ps,17,r.eref());}); updateTableState("contract-position", maxUpd(rows, LeitstandSourceRows.ContractPositionRow::upd), lastId(rows, LeitstandSourceRows.ContractPositionRow::dbk)); }
    private void syncCostUnits(TimeSyncRun run) { var rows = sourceClient.fetchCostUnits(watermark("cost-unit")); run.setRowsRead(run.getRowsRead()+rows.size()); batchUpsert("INSERT INTO TIME.ls_cost_unit(dbk, dbk_uid, upd, upd_anz, upd_uid, status, person_dbk, organization_dbk, contract_dbk, legacy_relation_4_dbk, legacy_relation_5_dbk, contract_position_dbk, project_name, project_id, project_task, mcl_id, mcl_name, mcl_desc, valid_from, valid_to, effort_plan) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (dbk) DO UPDATE SET dbk_uid=EXCLUDED.dbk_uid, upd=EXCLUDED.upd, upd_anz=EXCLUDED.upd_anz, upd_uid=EXCLUDED.upd_uid, status=EXCLUDED.status, person_dbk=EXCLUDED.person_dbk, organization_dbk=EXCLUDED.organization_dbk, contract_dbk=EXCLUDED.contract_dbk, legacy_relation_4_dbk=EXCLUDED.legacy_relation_4_dbk, legacy_relation_5_dbk=EXCLUDED.legacy_relation_5_dbk, contract_position_dbk=EXCLUDED.contract_position_dbk, project_name=EXCLUDED.project_name, project_id=EXCLUDED.project_id, project_task=EXCLUDED.project_task, mcl_id=EXCLUDED.mcl_id, mcl_name=EXCLUDED.mcl_name, mcl_desc=EXCLUDED.mcl_desc, valid_from=EXCLUDED.valid_from, valid_to=EXCLUDED.valid_to, effort_plan=EXCLUDED.effort_plan, updated_at=CURRENT_TIMESTAMP", rows, (ps,r)->{set(ps,1,r.dbk());set(ps,2,r.dbkUid());set(ps,3,r.upd());set(ps,4,r.updAnz());set(ps,5,r.updUid());set(ps,6,r.status());set(ps,7,r.personDbk());set(ps,8,r.organizationDbk());set(ps,9,r.contractDbk());set(ps,10,r.legacyRelation4Dbk());set(ps,11,r.legacyRelation5Dbk());set(ps,12,r.contractPositionDbk());set(ps,13,r.projectName());set(ps,14,r.projectId());set(ps,15,r.projectTask());set(ps,16,r.mclId());set(ps,17,r.mclName());set(ps,18,r.mclDesc());set(ps,19,ts(r.validFrom()));set(ps,20,ts(r.validTo()));set(ps,21,r.effortPlan());}); updateTableState("cost-unit", maxUpd(rows, LeitstandSourceRows.CostUnitRow::upd), lastId(rows, LeitstandSourceRows.CostUnitRow::dbk)); }
    private void syncTasks(TimeSyncRun run) { var rows = sourceClient.fetchTasks(watermark("task")); run.setRowsRead(run.getRowsRead()+rows.size()); batchUpsert("INSERT INTO TIME.ls_task(dbk, dbk_uid, upd, upd_anz, upd_uid, status, primary_cost_unit_dbk, secondary_cost_unit_dbk, tertiary_cost_unit_dbk, parent_task_dbk, lfdnr, task_type, mcl_id, mcl_name, mcl_desc, valid_from, valid_to, effort_plan, last_work, completion_percent, done_date, remark, n_rid, legacy_90702_rid, amount, created_source_at, legacy_90700_rid, legacy_90699_rid, quantity) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (dbk) DO UPDATE SET dbk_uid=EXCLUDED.dbk_uid, upd=EXCLUDED.upd, upd_anz=EXCLUDED.upd_anz, upd_uid=EXCLUDED.upd_uid, status=EXCLUDED.status, primary_cost_unit_dbk=EXCLUDED.primary_cost_unit_dbk, secondary_cost_unit_dbk=EXCLUDED.secondary_cost_unit_dbk, tertiary_cost_unit_dbk=EXCLUDED.tertiary_cost_unit_dbk, parent_task_dbk=EXCLUDED.parent_task_dbk, lfdnr=EXCLUDED.lfdnr, task_type=EXCLUDED.task_type, mcl_id=EXCLUDED.mcl_id, mcl_name=EXCLUDED.mcl_name, mcl_desc=EXCLUDED.mcl_desc, valid_from=EXCLUDED.valid_from, valid_to=EXCLUDED.valid_to, effort_plan=EXCLUDED.effort_plan, last_work=EXCLUDED.last_work, completion_percent=EXCLUDED.completion_percent, done_date=EXCLUDED.done_date, remark=EXCLUDED.remark, n_rid=EXCLUDED.n_rid, legacy_90702_rid=EXCLUDED.legacy_90702_rid, amount=EXCLUDED.amount, created_source_at=EXCLUDED.created_source_at, legacy_90700_rid=EXCLUDED.legacy_90700_rid, legacy_90699_rid=EXCLUDED.legacy_90699_rid, quantity=EXCLUDED.quantity, updated_at=CURRENT_TIMESTAMP", rows, (ps,r)->{set(ps,1,r.dbk());set(ps,2,r.dbkUid());set(ps,3,r.upd());set(ps,4,r.updAnz());set(ps,5,r.updUid());set(ps,6,r.status());set(ps,7,r.primaryCostUnitDbk());set(ps,8,r.secondaryCostUnitDbk());set(ps,9,r.tertiaryCostUnitDbk());set(ps,10,r.parentTaskDbk());set(ps,11,r.lfdnr());set(ps,12,r.taskType());set(ps,13,r.mclId());set(ps,14,r.mclName());set(ps,15,r.mclDesc());set(ps,16,ts(r.validFrom()));set(ps,17,ts(r.validTo()));set(ps,18,r.effortPlan());set(ps,19,ts(r.lastWork()));set(ps,20,r.completionPercent());set(ps,21,ts(r.doneDate()));set(ps,22,r.remark());set(ps,23,r.nRid());set(ps,24,r.legacy90702Rid());set(ps,25,r.amount());set(ps,26,ts(r.createdSourceAt()));set(ps,27,r.legacy90700Rid());set(ps,28,r.legacy90699Rid());set(ps,29,r.quantity());}); updateTableState("task", maxUpd(rows, LeitstandSourceRows.TaskRow::upd), lastId(rows, LeitstandSourceRows.TaskRow::dbk)); }
    private void syncPersonTaskAssignments(TimeSyncRun run) { var rows = sourceClient.fetchPersonTaskAssignments(watermark("person-task-assignment")); run.setRowsRead(run.getRowsRead()+rows.size()); batchUpsert("INSERT INTO TIME.ls_person_task_assignment(dbk, dbk_uid, upd, upd_anz, upd_uid, status, task_dbk, person_dbk, cost_unit_dbk, rtype, last_work, effort_plan) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (dbk) DO UPDATE SET dbk_uid=EXCLUDED.dbk_uid, upd=EXCLUDED.upd, upd_anz=EXCLUDED.upd_anz, upd_uid=EXCLUDED.upd_uid, status=EXCLUDED.status, task_dbk=EXCLUDED.task_dbk, person_dbk=EXCLUDED.person_dbk, cost_unit_dbk=EXCLUDED.cost_unit_dbk, rtype=EXCLUDED.rtype, last_work=EXCLUDED.last_work, effort_plan=EXCLUDED.effort_plan, updated_at=CURRENT_TIMESTAMP", rows, (ps,r)->{set(ps,1,r.dbk());set(ps,2,r.dbkUid());set(ps,3,r.upd());set(ps,4,r.updAnz());set(ps,5,r.updUid());set(ps,6,r.status());set(ps,7,r.taskDbk());set(ps,8,r.personDbk());set(ps,9,r.costUnitDbk());set(ps,10,r.rtype());set(ps,11,ts(r.lastWork()));set(ps,12,r.effortPlan());}); updateTableState("person-task-assignment", maxUpd(rows, LeitstandSourceRows.PersonTaskAssignmentRow::upd), lastId(rows, LeitstandSourceRows.PersonTaskAssignmentRow::dbk)); }
    private List<LeitstandSourceRows.TimeRecordingRow> syncTimeRecordings(TimeSyncRun run) { var rows = sourceClient.fetchTimeRecordings(watermark("time-recording")); run.setRowsRead(run.getRowsRead()+rows.size()); batchUpsert("INSERT INTO TIME.ls_time_recording(dbk, dbk_uid, upd, upd_anz, upd_uid, status, person_dbk, lfdnr, record_type, mcl_id, mcl_desc, recorded_from, recorded_to, effort, remark, url, activity_type_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (dbk) DO UPDATE SET dbk_uid=EXCLUDED.dbk_uid, upd=EXCLUDED.upd, upd_anz=EXCLUDED.upd_anz, upd_uid=EXCLUDED.upd_uid, status=EXCLUDED.status, person_dbk=EXCLUDED.person_dbk, lfdnr=EXCLUDED.lfdnr, record_type=EXCLUDED.record_type, mcl_id=EXCLUDED.mcl_id, mcl_desc=EXCLUDED.mcl_desc, recorded_from=EXCLUDED.recorded_from, recorded_to=EXCLUDED.recorded_to, effort=EXCLUDED.effort, remark=EXCLUDED.remark, url=EXCLUDED.url, activity_type_id=EXCLUDED.activity_type_id, updated_at=CURRENT_TIMESTAMP", rows, (ps,r)->{set(ps,1,r.dbk());set(ps,2,r.dbkUid());set(ps,3,r.upd());set(ps,4,r.updAnz());set(ps,5,r.updUid());set(ps,6,r.status());set(ps,7,r.personDbk());set(ps,8,r.lfdnr());set(ps,9,r.recordType());set(ps,10,r.mclId());set(ps,11,r.mclDesc());set(ps,12,ts(r.recordedFrom()));set(ps,13,ts(r.recordedTo()));set(ps,14,r.effort());set(ps,15,r.remark());set(ps,16,r.url());set(ps,17,r.activityTypeId());}); updateTableState("time-recording", maxUpd(rows, LeitstandSourceRows.TimeRecordingRow::upd), lastId(rows, LeitstandSourceRows.TimeRecordingRow::dbk)); return rows; }
    private void syncTimeRecordingAssignments(TimeSyncRun run) { var rows = sourceClient.fetchTimeRecordingAssignments(watermark("time-recording-assignment")); run.setRowsRead(run.getRowsRead()+rows.size()); batchUpsert("INSERT INTO TIME.ls_time_recording_assignment(dbk, dbk_uid, upd, upd_anz, upd_uid, status, person_task_assignment_dbk, time_recording_dbk, mcl_desc, effort, remark) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (dbk) DO UPDATE SET dbk_uid=EXCLUDED.dbk_uid, upd=EXCLUDED.upd, upd_anz=EXCLUDED.upd_anz, upd_uid=EXCLUDED.upd_uid, status=EXCLUDED.status, person_task_assignment_dbk=EXCLUDED.person_task_assignment_dbk, time_recording_dbk=EXCLUDED.time_recording_dbk, mcl_desc=EXCLUDED.mcl_desc, effort=EXCLUDED.effort, remark=EXCLUDED.remark, updated_at=CURRENT_TIMESTAMP", rows, (ps,r)->{set(ps,1,r.dbk());set(ps,2,r.dbkUid());set(ps,3,r.upd());set(ps,4,r.updAnz());set(ps,5,r.updUid());set(ps,6,r.status());set(ps,7,r.personTaskAssignmentDbk());set(ps,8,r.timeRecordingDbk());set(ps,9,r.mclDesc());set(ps,10,r.effort());set(ps,11,r.remark());}); updateTableState("time-recording-assignment", maxUpd(rows, LeitstandSourceRows.TimeRecordingAssignmentRow::upd), lastId(rows, LeitstandSourceRows.TimeRecordingAssignmentRow::dbk)); }

    public int materializeCanonicalTimeEntriesForPersonDbk(String personDbk, boolean rebuildProjection) {
        if (personDbk == null || personDbk.isBlank()) {
            throw new IllegalArgumentException("personDbk must not be blank");
        }
        List<LeitstandTimeRecording> recordings = timeRecordingRepository.findByPersonDbkOrderByRecordedFromAscDbkAsc(personDbk);
        if (recordings.isEmpty()) {
            log.info("No Leitstand time recordings found for personDbk={}", personDbk);
            return 0;
        }
        //upsertCanonicalTimeEntriesForImportedRecordings(recordings);
        if (rebuildProjection && properties.getLeitstand().isBuildSearchProjection()) {
            projectionService.refreshForPersonDbk(personDbk);
        }
        return recordings.size();
    }

    @Transactional
    public int materializeCanonicalTimeEntriesForPersonNumber(Integer personNumber, boolean rebuildProjection) {
        if (personNumber == null) {
            throw new IllegalArgumentException("personNumber must not be null");
        }
        LeitstandPerson person = personRepository.findByPersonNumber(personNumber)
                .orElseThrow(() -> new IllegalArgumentException("Unknown Leitstand person number: " + personNumber));
        return materializeCanonicalTimeEntriesForPersonDbk(person.getDbk(), rebuildProjection);
    }

    @Transactional
    protected void upsertCanonicalTimeEntriesForImportedRecordings(List<LeitstandTimeRecording> recordings) {
        Map<String, String> personNames = loadPersonNames();
        int count = 0;
        for (LeitstandTimeRecording recording : recordings) {
            String businessKey = "TIME:LEITSTAND:" + recording.getDbk();
            Document document = documentRepository.findByBusinessKey(businessKey).orElseGet(() -> Document.builder().businessKey(businessKey).documentType(DocumentType.TIME_ENTRY).documentFamily(DocumentFamily.TIME).visibility(DocumentVisibility.PUBLIC).status(DocumentStatus.RECEIVED).build());
            document.setTitle(buildTitle(new LeitstandSourceRows.TimeRecordingRow(
                    recording.getDbk(), recording.getDbkUid(), recording.getUpd(), recording.getUpdAnz(), recording.getUpdUid(), recording.getStatus(), recording.getPersonDbk(), recording.getLfdnr(), recording.getRecordType(), recording.getMclId(), recording.getMclDesc(), recording.getRecordedFrom(), recording.getRecordedTo(), recording.getEffort(), recording.getRemark(), recording.getUrl(), recording.getActivityTypeId()
            ), personNames.get(recording.getPersonDbk())));
            document.setSummary(buildSummary(new LeitstandSourceRows.TimeRecordingRow(
                    recording.getDbk(), recording.getDbkUid(), recording.getUpd(), recording.getUpdAnz(), recording.getUpdUid(), recording.getStatus(), recording.getPersonDbk(), recording.getLfdnr(), recording.getRecordType(), recording.getMclId(), recording.getMclDesc(), recording.getRecordedFrom(), recording.getRecordedTo(), recording.getEffort(), recording.getRemark(), recording.getUrl(), recording.getActivityTypeId()
            )));
            document = documentRepository.save(document);

            Document finalDocument = document;
            TimeEntry entry = timeEntryRepository.findBySourceSystemAndExternalId(TimeSourceSystem.LEITSTAND, recording.getDbk()).orElseGet(() -> TimeEntry.builder().sourceSystem(TimeSourceSystem.LEITSTAND).externalId(recording.getDbk()).document(finalDocument).build());
            entry.setDocument(document);
            entry.setPersonExternalId(recording.getPersonDbk());
            entry.setPersonDisplayName(personNames.get(recording.getPersonDbk()));
            entry.setEntryStart(recording.getRecordedFrom());
            entry.setEntryEnd(recording.getRecordedTo());
            entry.setDurationSeconds(toSeconds(recording.getEffort()));
            entry.setDescriptionShort(recording.getMclDesc());
            entry.setDescriptionLong(recording.getRemark());
            entry.setRawStatus(recording.getStatus() == null ? null : String.valueOf(recording.getStatus()));
            entry.setSearchAnchorLabel(document.getTitle());
            entry = timeEntryRepository.save(entry);
            timeEntryRepository.flush();

            log.info("Leitstand time recordings businessKey={}, count={}", businessKey, ++count);

            jdbcTemplate.update("""
                update TIME.ls_time_recording set time_entry_id = ? where dbk = ?
            """, entry.getId(), recording.getDbk());
            upsertSourceLink(entry, "TIME_RECORDING", recording.getDbk(), null);
            if (recording.getPersonDbk() != null) upsertSourceLink(entry, "PERSON", recording.getPersonDbk(), recording.getDbk());
            if (recording.getActivityTypeId() != null) upsertSourceLink(entry, "ACTIVITY_TYPE", String.valueOf(recording.getActivityTypeId()), recording.getDbk());
        }
    }

    //@Transactional
    protected void upsertCanonicalTimeEntries(List<LeitstandSourceRows.TimeRecordingRow> rows) {
        Map<String, String> personNames = loadPersonNames();
        for (var row : rows) {
            String businessKey = "TIME:LEITSTAND:" + row.dbk();
            Document document = documentRepository.findByBusinessKey(businessKey).orElseGet(() -> Document.builder().businessKey(businessKey).documentType(DocumentType.TIME_ENTRY).documentFamily(DocumentFamily.TIME).visibility(DocumentVisibility.PUBLIC).status(DocumentStatus.RECEIVED).build());
            document.setTitle(buildTitle(row, personNames.get(row.personDbk())));
            document.setSummary(buildSummary(row));
            document = documentRepository.save(document);

            Document finalDocument = document;
            TimeEntry entry = timeEntryRepository.findBySourceSystemAndExternalId(TimeSourceSystem.LEITSTAND, row.dbk()).orElseGet(() -> TimeEntry.builder().sourceSystem(TimeSourceSystem.LEITSTAND).externalId(row.dbk()).document(finalDocument).build());
            entry.setDocument(document);
            entry.setPersonExternalId(row.personDbk());
            entry.setPersonDisplayName(personNames.get(row.personDbk()));
            entry.setEntryStart(row.recordedFrom());
            entry.setEntryEnd(row.recordedTo());
            entry.setDurationSeconds(toSeconds(row.effort()));
            entry.setDescriptionShort(row.mclDesc());
            entry.setDescriptionLong(row.remark());
            entry.setRawStatus(row.status() == null ? null : String.valueOf(row.status()));
            entry.setSearchAnchorLabel(document.getTitle());
            entry = timeEntryRepository.save(entry);
            timeEntryRepository.flush();

            jdbcTemplate.update("""
                update TIME.ls_time_recording set time_entry_id = ? where dbk = ?
            """, entry.getId(), row.dbk());
            upsertSourceLink(entry, "TIME_RECORDING", row.dbk(), null);
            if (row.personDbk() != null) upsertSourceLink(entry, "PERSON", row.personDbk(), row.dbk());
            if (row.activityTypeId() != null) upsertSourceLink(entry, "ACTIVITY_TYPE", String.valueOf(row.activityTypeId()), row.dbk());
        }
    }

    private void upsertSourceLink(TimeEntry entry, String type, String externalId, String parentExternalId) {
        TimeEntrySourceLink link = sourceLinkRepository.findByTimeEntry_Id(entry.getId()).stream()
                .filter(it -> it.getSourceSystem() == TimeSourceSystem.LEITSTAND && type.equals(it.getSourceEntityType()) && externalId.equals(it.getSourceExternalId()))
                .findFirst()
                .orElseGet(() -> TimeEntrySourceLink.builder().timeEntry(entry).sourceSystem(TimeSourceSystem.LEITSTAND).sourceEntityType(type).sourceExternalId(externalId).build());
        link.setParentSourceExternalId(parentExternalId);
        sourceLinkRepository.save(link);
    }

    private Map<String, String> loadPersonNames() {
        Map<String, String> result = new HashMap<>();
        jdbcTemplate.query("""
                select dbk, first_name, last_name from TIME.ls_person
            """, rs -> {
            result.put(rs.getString("dbk"), firstNonBlank(joinName(rs.getString("first_name"), rs.getString("last_name")), rs.getString("last_name"), rs.getString("dbk")));
        });
        return result;
    }

    private String buildTitle(LeitstandSourceRows.TimeRecordingRow row, String personName) {
        String base = firstNonBlank(row.mclDesc(), row.remark(), row.mclId());
        if (base == null) base = "Time entry " + row.dbk();
        return personName == null ? base : personName + " - " + base;
    }

    private String buildSummary(LeitstandSourceRows.TimeRecordingRow row) {
        StringBuilder sb = new StringBuilder();
        if (row.mclDesc() != null && !row.mclDesc().isBlank()) sb.append(row.mclDesc().trim());
        if (row.remark() != null && !row.remark().isBlank()) { if (sb.length() > 0) sb.append(" | "); sb.append(row.remark().trim()); }
        if (row.mclId() != null && !row.mclId().isBlank()) { if (sb.length() > 0) sb.append(" | "); sb.append("ID ").append(row.mclId().trim()); }
        return sb.length() == 0 ? null : sb.toString();
    }

    private Long toSeconds(BigDecimal effort) { return effort == null ? null : effort.multiply(BigDecimal.valueOf(3600)).setScale(0, RoundingMode.HALF_UP).longValue(); }
    private String joinName(String first, String last) { return ((first == null ? "" : first.trim()) + (last == null ? "" : " " + last.trim())).trim(); }
    private String firstNonBlank(String... vals) { for (String v : vals) if (v != null && !v.trim().isEmpty()) return v.trim(); return null; }
    private String watermark(String scope) { return properties.getLeitstand().isIncrementalEnabled() ? syncStateRepository.findBySourceSystemAndScopeKey(TimeSourceSystem.LEITSTAND, tableScope(scope)).map(TimeSyncState::getLastSeenWatermark).orElse(null) : null; }
    private String tableScope(String scope) { return "leitstand:" + scope; }
    private <T> String lastId(List<T> rows, java.util.function.Function<T,String> id) { return rows.isEmpty() ? null : id.apply(rows.get(rows.size()-1)); }
    private <T> String maxUpd(List<T> rows, java.util.function.Function<T,String> upd) { return rows.stream().map(upd).filter(v -> v != null && !v.isBlank()).max(String::compareTo).orElse(null); }

    @Transactional
    protected void updateTableState(String scope, String watermark, String lastExternalId) {
        TimeSyncState state = syncStateRepository.findBySourceSystemAndScopeKey(TimeSourceSystem.LEITSTAND, tableScope(scope)).orElseGet(() -> TimeSyncState.builder().sourceSystem(TimeSourceSystem.LEITSTAND).scopeKey(tableScope(scope)).build());
        state.setLastAttemptedSyncAt(OffsetDateTime.now());
        state.setLastSuccessfulSyncAt(OffsetDateTime.now());
        state.setLastSeenWatermark(watermark);
        state.setLastSeenExternalId(lastExternalId);
        syncStateRepository.save(state);
    }

    @Transactional
    protected void updateGlobalState(TimeSyncRun run, String notes) {
        TimeSyncState state = syncStateRepository.findBySourceSystemAndScopeKey(TimeSourceSystem.LEITSTAND, GLOBAL_SCOPE).orElseGet(() -> TimeSyncState.builder().sourceSystem(TimeSourceSystem.LEITSTAND).scopeKey(GLOBAL_SCOPE).build());
        state.setLastAttemptedSyncAt(run.getStartedAt());
        if (run.getStatus() == TimeSyncRunStatus.COMPLETED) state.setLastSuccessfulSyncAt(run.getFinishedAt());
        state.setLastRun(run);
        state.setNotes(notes);
        syncStateRepository.save(state);
    }

    private <T> void batchUpsert(String sql, List<T> rows, Binder<T> binder) {
        jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() {
            @Override public void setValues(PreparedStatement ps, int i) throws java.sql.SQLException { binder.bind(ps, rows.get(i)); }
            @Override public int getBatchSize() { return rows.size(); }
        });
    }

    private void set(PreparedStatement ps, int index, Object value) throws java.sql.SQLException { ps.setObject(index, value); }
    private Timestamp ts(OffsetDateTime t) { return t == null ? null : Timestamp.from(t.toInstant()); }

    @FunctionalInterface
    private interface Binder<T> { void bind(PreparedStatement ps, T row) throws java.sql.SQLException; }
}
