# include "dvomerge.h"
# define DEBUG 1

// in parallel mode, the target database may be distributed, while the input database is
// only on a single machine

// this function only merges tables which are at equal or shallower depth compared to the
// output tables

int dvomergeUpdate_parallel (char *input, char *output, SkyTable *outsky, IDmapType *IDmap);

int dvomergeUpdate_catalogs (char *input, char *output, SkyTable *outsky, SkyList *inlist, int NsecfiltInput, int NsecfiltOutput, IDmapType *IDmap, int *secfiltMap) {

  off_t i, j;
  SkyList *outlist;
  Catalog incatalog, outcatalog;

  if (PARALLEL && !HOST_ID) {
    int status = dvomergeUpdate_parallel (input, output, outsky, IDmap);
    return status;
  }

  // load the list of hosts
  HostTable *table_input = NULL;
  if (PARALLEL_INPUT) {
    table_input = HostTableLoad (input, inlist->hosts);
    if (!table_input) {
      fprintf (stderr, "ERROR: failure reading Host Table %s for database %s\n", inlist->hosts, input);
      exit (1);
    }    

    // ensure we have absolute paths for hostdirs
    for (i = 0; i < table_input->Nhosts; i++) {
      char *tmppath = abspath (table_input->hosts[i].pathname, DVO_MAX_PATH);
      free (table_input->hosts[i].pathname);
      table_input->hosts[i].pathname = tmppath;
    }
  }

  // loop over the populated input regions
  for (i = 0; i < inlist[0].Nregions; i++) {
    if (!inlist[0].regions[i][0].table) continue;
    if (VERBOSE) fprintf (stderr, "input: %s\n", inlist[0].regions[i][0].name);

    // in a parallel context, we need to re-map the input filename
    char *filename_input = strcreate (inlist[0].filename[i]);
    if (PARALLEL_INPUT) {

      int hostID_input = inlist[0].regions[i]->hostID;
      int table_index = table_input->index[hostID_input];
      char *hostdir_input = table_input->hosts[table_index].pathname;

      // set the parameters which guide catalog open/load/create
      char hostfile_input[DVO_MAX_PATH];
      snprintf (hostfile_input, DVO_MAX_PATH, "%s/%s.cpt", hostdir_input, inlist[0].regions[i]->name);
      free (filename_input);
      filename_input = strcreate (hostfile_input);
    }

    // SkyListByBounds will return neighbor catalogs if the boundaries exactly match (due to rounding).  Since the regions are not infinitely small, 
    // compare to a slightly reduced footprint
    float dPos = 2.0/3600.0;
    outlist = SkyListByBounds (outsky, -1, inlist[0].regions[i][0].Rmin + dPos, inlist[0].regions[i][0].Rmax - dPos, inlist[0].regions[i][0].Dmin + dPos, inlist[0].regions[i][0].Dmax - dPos);

    OutputStatus *outstat = OutputStatusInit (outlist->Nregions);

    // there may be more than one output region for a given input region.  
    // are any of these output regions relevant to this machine?
    int found = FALSE;
    for (j = 0; j < outlist[0].Nregions; j++) {
      outstat[j].valid = HostTableTestHost(outlist[0].regions[j], HOST_ID);
      found = (found || outstat[j].valid);
    }

    // skip this input table for if no output files are on this machine
    if (!found) {
      OutputStatusFree (outstat, outlist->Nregions);
      SkyListFree (outlist); 
      free (filename_input);
      continue; 
    }

    // get stats for history check, skip input catalog if file not found (NULL inStats)
    dmhObjectStats *inStats = dmhObjectStatsRead (filename_input);
    if (!inStats) {
      if (VERBOSE) fprintf (stderr, "skipping %s, empty \n", filename_input);
      OutputStatusFree (outstat, outlist->Nregions);
      SkyListFree (outlist); 
      free (filename_input);
      continue;
    }

    // Check if any of the output files have NOT yet received data from this input file
    // If none have been missed, we can skip the input file completely
    int missed = FALSE;
    for (j = 0; j < outlist[0].Nregions; j++) {
      if (!outstat[j].valid) continue;

      // set the parameters which guide catalog open/load/create
      char hostfile[DVO_MAX_PATH];
      snprintf (hostfile, DVO_MAX_PATH, "%s/%s.cpt", HOSTDIR, outlist[0].regions[j]->name);
      char *filename = HOST_ID ? hostfile : outlist[0].filename[j];
      outstat[j].filename = strcreate (filename);

      outstat[j].history = dmhObjectRead (outstat[j].filename);

      // have we already merged this database?
      outstat[j].missed = !dmhObjectCheck (outstat[j].history, inStats);
      missed = (missed || outstat[j].missed);
    }
    if (!FORCE_MERGE && !missed) {
      if (VERBOSE || VERIFY) fprintf (stderr, "skipping %s, already merged\n", filename_input);
      OutputStatusFree (outstat, outlist->Nregions);
      dmhObjectStatsFree (inStats);
      SkyListFree (outlist); 
      free (filename_input);
      continue;
    }
    if (VERIFY) { 
      fprintf (stderr, "%s NOT merged\n", filename_input);
      OutputStatusFree (outstat, outlist->Nregions);
      dmhObjectStatsFree (inStats);
      SkyListFree (outlist); 
      free (filename_input);
      continue;
    }

    // read the input catalog
    LoadCatalog (&incatalog, &inlist[0].regions[i][0], filename_input, "r", NsecfiltInput);

    // skip empty input catalogs
    if (!incatalog.Naves_disk) {
	dvo_catalog_unlock (&incatalog);
	dvo_catalog_free (&incatalog);
	OutputStatusFree (outstat, outlist->Nregions);
	dmhObjectStatsFree (inStats);
	SkyListFree (outlist); 
	free (filename_input);
	continue;
    }

    if (!incatalog.sorted) {
      fprintf (stderr, "ERROR: input catalog %s is not sorted (and must be for dvomerge)\n", filename_input);
      exit (1);
    }

    dvo_update_image_IDs (IDmap, &incatalog);

    // merge input into the appropriate output tables
    for (j = 0; j < outlist[0].Nregions; j++) {
      // skip tables for which the output files are not on this machine
      if (!outstat[j].valid) continue; 

      // skip if we have already done the merge
      if (!FORCE_MERGE && !outstat[j].missed) continue; 

      if (VERBOSE) fprintf (stderr, "output : %s\n", outlist[0].regions[j][0].name);

      // the real filename
      outcatalog.filename = outstat[j].filename; 

      // load input catalog
      LoadCatalog (&outcatalog, outlist[0].regions[j], outcatalog.filename, "w", NsecfiltOutput);

      // if no catalog already exists, use the input catalog to define the format
      if (outcatalog.Naves_disk == 0) {
	outcatalog.catformat = incatalog.catformat;
      }

      merge_catalogs_old (outlist[0].regions[j], &outcatalog, &incatalog, RADIUS, secfiltMap);

      if (outstat[j].missed) {
	dmhObjectAdd (outstat[j].history, &outcatalog.header, inStats);
      }

      if (!dvo_catalog_backup (&outcatalog, TRUE)) {
	fprintf (stderr, "ERROR: failed to make backup for catalog %s\n", outlist[0].filename[j]);
	exit (1);
      }

      // if we receive a signal which would cause us to exit, wait until the full catalog is written
      SetProtect (TRUE);
      if (!dvo_catalog_save (&outcatalog, VERBOSE)) {
	fprintf (stderr, "ERROR: failed to save catalog %s\n", outlist[0].filename[j]);
	exit (1);
      }
      SetProtect (FALSE);

      if (!dvo_catalog_unlink_backup (&outcatalog, TRUE)) {
	fprintf (stderr, "WARNING: failed to remove backup for catalog %s\n", outlist[0].filename[j]);
      }

      dvo_catalog_unlock (&outcatalog);
      dvo_catalog_free (&outcatalog);

      fprintf (stderr, "merged %s into %s\n", inlist[0].regions[i][0].name, outlist[0].regions[j][0].name);
    }

    OutputStatusFree (outstat, outlist->Nregions);
    SkyListFree (outlist);
    dmhObjectStatsFree (inStats);

    free (filename_input);
    dvo_catalog_unlock (&incatalog);
    dvo_catalog_free (&incatalog);
  }
  return TRUE;
}

// launch the dvomergeUpdate_client jobs to the parallel hosts
int dvomergeUpdate_parallel (char *input, char *output, SkyTable *outsky, IDmapType *IDmap) {

  // ensure that the paths are absolute path names
  char *absinput  = abspath (input,  DVO_MAX_PATH);
  char *absoutput = abspath (output, DVO_MAX_PATH);

  // name of image subset file:
  char IDmapFilename[DVO_MAX_PATH];
  snprintf (IDmapFilename, DVO_MAX_PATH, "%s/IDmap.fits", absoutput);

  if (!VERIFY_CATALOG_ONLY) {
    // save IDmap information
    if (!IDmapSave (IDmapFilename, IDmap)) {
      fprintf (stderr, "ERROR: failure to save the image ID map\n");
      exit (1);
    }
  }

  // load the list of hosts
  HostTable *table = HostTableLoad (output, outsky->hosts);
  if (!table) {
    fprintf (stderr, "ERROR: failure reading Host Table %s for database %s\n", outsky->hosts, output);
    exit (1);
  }    

  int i;
  for (i = 0; i < table->Nhosts; i++) {

    // ensure that the paths are absolute path names
    char *tmppath = abspath (table->hosts[i].pathname, DVO_MAX_PATH);
    free (table->hosts[i].pathname);
    table->hosts[i].pathname = tmppath;

    // options / arguments that can affect relastro_client -update-objects:
    char command[DVO_MAX_PATH];
    snprintf (command, DVO_MAX_PATH, "dvomerge_client %s into %s -hostID %d -hostdir %s -region %f %f %f %f -D ADDSTAR_RADIUS %f", 
	      absinput, absoutput, table->hosts[i].hostID, table->hosts[i].pathname, 
	      UserPatch.Rmin, UserPatch.Rmax, UserPatch.Dmin, UserPatch.Dmax, RADIUS
      );

    char tmpline[DVO_MAX_PATH];
    if (VERBOSE)             { snprintf (tmpline, DVO_MAX_PATH, "%s -v",               command); strcpy (command, tmpline); }
    if (VERIFY)              { snprintf (tmpline, DVO_MAX_PATH, "%s -verify",          command); strcpy (command, tmpline); }
    if (VERIFY_CATALOG_ONLY) { snprintf (tmpline, DVO_MAX_PATH, "%s -verify-catalogs", command); strcpy (command, tmpline); }
    if (REPLACE_BY_PHOTCODE) { snprintf (tmpline, DVO_MAX_PATH, "%s -replace",         command); strcpy (command, tmpline); }
    if (PARALLEL_INPUT)      { snprintf (tmpline, DVO_MAX_PATH, "%s -parallel-input",  command); strcpy (command, tmpline); }
    if (FORCE_MERGE)         { snprintf (tmpline, DVO_MAX_PATH, "%s -force-merge",     command); strcpy (command, tmpline); }

    // add some config variables:
    snprintf (tmpline, DVO_MAX_PATH, "%s -D CATMODE %s",    command, CATMODE);   strcpy (command, tmpline);
    snprintf (tmpline, DVO_MAX_PATH, "%s -D CATFORMAT %s",  command, CATFORMAT); strcpy (command, tmpline);
    snprintf (tmpline, DVO_MAX_PATH, "%s -D SKY_DEPTH %d",  command, SKY_DEPTH); strcpy (command, tmpline);

    fprintf (stderr, "command: %s\n", command);

    if (PARALLEL_MANUAL) continue;

    if (PARALLEL_SERIAL) {
      int status = system (command);
      if (status) {
	fprintf (stderr, "ERROR running photdbc_client\n");
	exit (2);
      }
    } else {
      // launch the job on the remote machine (no handshake)
      int errorInfo = 0;
      int pid = rconnect ("ssh", table->hosts[i].hostname, command, table->hosts[i].stdio, &errorInfo, FALSE);
      if (!pid) {
	if (DEBUG) fprintf (stderr, "failure to start %s (error %d)\n", table->hosts[i].hostname, errorInfo);
	exit (1);
      }
      table->hosts[i].pid = pid; // save for future reference
    }
  }

  if (PARALLEL_MANUAL) {
    fprintf (stderr, "run the photdbc_client commands above.  when these are done, hit return\n");
    getchar();
  }
  if (!PARALLEL_MANUAL && !PARALLEL_SERIAL) {
    int status = HostTableWaitJobsGetIO (table, __FILE__, __LINE__, VERBOSE);
    if (!status) {
      fprintf (stderr, "error running one of the remote clients\n");
      return status;
    }
  }

  return TRUE;
}      
