*! _diddesign_parallel_worker.ado - Parallel bootstrap worker program
*!
*! Worker program executed inside each parallel sub-process.
*! Called via: parallel, by(_ddpar_worker_id): do _diddesign_parallel_worker.ado
*! Each invocation receives a 1-row dataset with the worker's iteration parameters.

version 16.0

// Read worker parameters from the 1-row dispatch dataset
// (supplied by parallel, by(_ddpar_worker_id) from the auxiliary dispatch dataset)
local b_start = _ddpar_bstart[1]
local b_end   = _ddpar_bend[1]
local seed_k  = _ddpar_seed[1]
local k       = _ddpar_worker_id[1]
local design  = "$DDPAR_DESIGN"
local outfile = "${DDPAR_TMPDIR}/boot_chunk_`k'.dta"

// Load the full working dataset and Mata library
use "${DDPAR_DATA}", clear
quietly do "${DDPAR_MATALIB}"

// Data preparation: rebuild did_data and (for SA) did_option Mata structs
// by calling the same prep functions used by the main process.
// Workers receive the ALREADY-PREPARED dataset (with derived variables
// generated by _diddesign_prep in the main process). They only need to
// populate the Mata structs from the existing Stata variables, NOT re-run
// _diddesign_prep or fvexpand.

if ("`design'" == "did") {
    // Build Mata lead rowvector from DDPAR_LEAD global
    mata: _par_lead = strtoreal(tokens("${DDPAR_LEAD}"))

    // Populate external did_dat from the already-prepared Stata variables.
    mata: _diddesign_populate_data(                      ///
        "${DDPAR_OUTCOME}", "${DDPAR_TREATMENT}",        ///
        "${DDPAR_ID}", "${DDPAR_IDTIMEVAR}",             ///
        "${DDPAR_EXPANDED_COVARIATES}",                  ///
        "${DDPAR_CLUSTER}",                              ///
        "${DDPAR_GIVAR}", "${DDPAR_ITVAR}",             ///
        "${DDPAR_IDTIMESTDVAR}", "${DDPAR_DELTAVAR}",   ///
        strtoreal("${DDPAR_NOBS}"),                     ///
        strtoreal("${DDPAR_NUNITS}"),                   ///
        strtoreal("${DDPAR_NPERIODS}"),                 ///
        strtoreal("${DDPAR_TREATYEAR}"),                ///
        strtoreal("${DDPAR_ISPANEL}"),                  ///
        "${DDPAR_TOUSEVAR}")
}
else {
    // SA design: uses its own data preparation function
    mata: _did_sa_prepare_data(                          ///
        "${DDPAR_OUTCOME}", "${DDPAR_TREATMENT}",        ///
        "${DDPAR_ID}", "${DDPAR_TIME}",                  ///
        "${DDPAR_CLUSTER}",                              ///
        "${DDPAR_EXPANDED_COVARIATES}",                  ///
        "${DDPAR_TOUSEVAR}")

    // Populate did_option for sa_double_did() calls.
    mata: _diddesign_populate_option(                    ///
        strtoreal("${DDPAR_NBOOT}"),                     ///
        1,                                               ///
        strtoreal("${DDPAR_SEBOOT}"),                    ///
        "${DDPAR_CLUSTER}",                              ///
        strtoreal(tokens("${DDPAR_LEAD}")),              ///
        strtoreal("${DDPAR_THRES}"),                     ///
        J(1, 0, .),                                      ///
        strtoreal("${DDPAR_LEVEL}"),                     ///
        .)
    mata: did_opt.quiet = strtoreal("${DDPAR_QUIET}")
}

// Call Mata chunk function
// Note: the dispatch dataset uses missing (.) for "no seed", not -1.
// missing() must be tested before calling strofreal() to avoid spurious values.
local seed_mata = cond(missing(`seed_k'), ".", strofreal(`seed_k'))

if ("`design'" == "did") {
    mata: _ddw_chunk = did_boot_std_chunk(did_dat, _par_lead, `b_start', `b_end', ("`seed_mata'" == "." ? . : strtoreal("`seed_mata'")))
}
else {
    mata: _ddw_chunk = did_boot_sa_chunk(did_dat, did_opt, `b_start', `b_end', ("`seed_mata'" == "." ? . : strtoreal("`seed_mata'")))
}
mata: st_numscalar("_boot_chunk_nsuccess", _ddw_chunk.n_successful)
mata: st_numscalar("_boot_chunk_nfailed", _ddw_chunk.n_failed)
mata: st_matrix("_boot_chunk_est", _ddw_chunk.n_successful > 0 ? _ddw_chunk.estimates : J(0, cols(_ddw_chunk.estimates), .))

// Write result to output file
// svmat converts the Stata matrix to dataset columns named _boot_col1, _boot_col2, ...
// Each row is one successful bootstrap iteration; each column is one DID or sDID estimate.
// save emptyok ensures the file is created even when n_successful == 0.
local n_success = `=scalar(_boot_chunk_nsuccess)'
local n_failed  = `=scalar(_boot_chunk_nfailed)'

qui {
    preserve
    clear
    if `n_success' > 0 {
        svmat double _boot_chunk_est, name(_boot_col)
        gen byte _ddpar_is_result = 1
    }
    else {
        set obs 1
        gen byte _ddpar_is_result = 0
    }
    // Store failure count as a variable for coordinator to aggregate
    gen long _ddpar_n_failed = `n_failed'
    save "`outfile'", replace emptyok
    restore
}