# 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;
  }

  // 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);

    // 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);

    // 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; !found && (j < outlist[0].Nregions); j++) {
      found = (found || HostTableTestHost(outlist[0].regions[j], HOST_ID));
    }

    // skip this input table for if no output files are on this machine
    if (!found) {
      SkyListFree (outlist); 
      continue; 
    }

    // get the stats on this input file (for comparison with the output headers)
    struct stat instats;
    int stat_result = stat (inlist[0].filename[i], &instats);
    if (stat_result) {
      if (errno == ENOENT) continue;
      fprintf (stderr, "cannot read stats on input file %s\n", inlist[0].filename[i]);
      perror ("stats error message:");
      exit (2);
    }

    // instats.st_size & instats.st_mtime

    // check if any of the output files have NOT yet received data from this input file

    int missed = FALSE;
    for (j = 0; !missed && (j < outlist[0].Nregions); j++) {
      if (!HostTableTestHost(outlist[0].regions[j], HOST_ID)) 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];

      // get the stats on this input file (for comparison with the output headers)
      struct stat outstats;
      stat_result = stat (filename, &outstats);
      if (stat_result) {
	if (errno == ENOENT) {
	  missed = TRUE;
	  break;
	}
	fprintf (stderr, "cannot read stats on output file %s\n", filename);
	perror ("stats error message:");
	exit (2);
      }

      FILE *fout = fopen (filename, "r");
      if (!fout) {
	fprintf (stderr, "problem opening output file to read header %s\n", filename);
	perror ("stats error message:");
	exit (2);
      }

      Header outheader;
      if (!gfits_fread_header (fout, &outheader)) {
	fprintf (stderr, "problem reading header for output file %s\n", filename);
	exit (2);
      }
      
      fclose (fout);

      // update header of output catalog
      long long last_size;
      char last_moddate[80];
      gfits_scan (&outheader, "LMRG_SZ", "%lld", 1, &last_size);      
      gfits_scan (&outheader, "LMRG_DT", "%s", 1, last_moddate);      

      time_t last_mod = ohana_date_to_sec (last_moddate);

      if (last_size != instats.st_size) {
	missed = TRUE;
	break;
      }
      
      if (last_mod != instats.st_mtime) {
	missed = TRUE;
	break;
      }
    }
    if (!missed) {
      fprintf (stderr, "skipping %s, already merged\n", inlist[0].filename[i]);
      continue;
    }

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

    // skip empty input catalogs
    if (!incatalog.Naves_disk) {
	dvo_catalog_unlock (&incatalog);
	dvo_catalog_free (&incatalog);
	SkyListFree (outlist); 
	continue;
    }

    char *moddate = ohana_sec_to_date (instats.st_mtime);

    // 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 (!HostTableTestHost(outlist[0].regions[j], HOST_ID)) continue; 

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

      // 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);
      outcatalog.filename  = HOST_ID ? hostfile : outlist[0].filename[j];

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

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

      outcatalog.catflags = LOAD_AVES | LOAD_MEAS | LOAD_MISS | LOAD_SECF;

      // update header of output catalog
      long long last_size;
      char last_moddate[80];
      int status_size = gfits_scan (&outcatalog.header, "LMRG_SZ", "%lld", 1, &last_size);      
      int status_date = gfits_scan (&outcatalog.header, "LMRG_DT", "%s", 1, last_moddate);      

      // save the previous size of merged entry
      if (status_size && status_date) {
	gfits_modify (&outcatalog.header, "LMRG_SZ1", "%lld", 1, (long long) last_size);      
	gfits_modify (&outcatalog.header, "LMRG_DT1", "%s", 1, last_moddate);      
      }

      // update header of output catalog
      gfits_modify (&outcatalog.header, "LMRG_SZ", "%lld", 1, (long long) instats.st_size);      
      gfits_modify (&outcatalog.header, "LMRG_DT", "%s", 1, moddate);      

      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);
    }
    SkyListFree (outlist);
    free (moddate);

    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);

  // 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 (REPLACE_BY_PHOTCODE) { snprintf (tmpline, DVO_MAX_PATH, "%s -replace", command); 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) {
    HostTableWaitJobsGetIO (table, __FILE__, __LINE__, VERBOSE);
  }

  return TRUE;
}      
