/*
 * Copyright (C) 2007-2008 by CERN/IT/GD/ITR
 * All rights reserved
 */

#ifndef lint
static char sccsid[] = "@(#)$RCSfile: dpmcopy_push.c,v $ $Revision: 1.1 $ $Date: 2008/09/24 11:25:00 $ CERN IT-GD/ITR Jean-Philippe Baud";
#endif /* not lint */

/*      dpmcopy_push - copy files when source is local and target is remote */
#include <errno.h>
#include <sys/types.h>
#include <fcntl.h>
#include <stdlib.h>
#include <string.h>
#include "dpm.h"
#include "dpm_api.h"
#include "dpm_server.h"
#include "dpmcopy_server.h"
#include "serrno.h"
#define DEFPOLLINT 5
extern int nb_file_err;

int
dpmcopy_push (struct dpm_dbfd *dbfd, struct dpm_req *dpm_req, struct dpm_copy_filereq *cpr_entry, int nbfiles)
{
	char errbuf[256];
	char errstring[256];
	struct dpm_filestatus *filestatuses;
	int *filstatus = NULL;
	char func[16];
	struct dpm_getfilereq *getfilereqs;
	struct dpm_getfilestatus *getfilestatuses = NULL;
	int i;
	int j = 0;
	int nbgetreplies;
	int nbputreplies;
	int nbreplies;
	char *p;
	static char *protocols[] = {"gsiftp"};
	struct dpm_putfilereq *putfilereqs;
	struct dpm_putfilestatus *putfilestatus = NULL;
	char *q;
	int r = 0;
	int r1;
	char r_tokeng[CA_MAXDPMTOKENLEN+1];
	char *r_tokenp;
	dpm_dbrec_addr rec_addrf;
	char srm_endpoint[256];
	int status;
	int statusg;

	strcpy (func, "dpmcopy_push");
	if (nbfiles == 0)
		return (0);	/* nothing to be done */

	nb_file_err = 0;

	/* build the get/put requests */

	if ((getfilereqs = calloc (nbfiles, sizeof(struct dpm_getfilereq))) == NULL ||
	    (putfilereqs = calloc (nbfiles, sizeof(struct dpm_putfilereq))) == NULL ||
	    (filstatus = calloc (nbfiles, sizeof(int))) == NULL) {
		dpm_req->status = DPM_FAILED | ENOMEM;
		return (-1);
	}
	for (i = 0; i < dpm_req->nbreqfiles; i++) {
		(void) dpm_start_tr (0, dbfd);
		if (dpm_get_cpr_by_fullid (dbfd, dpm_req->r_token, i,
		    cpr_entry, 1, &rec_addrf) < 0) {
			dpm_abort_tr (dbfd);
			continue;
		}
		if (cpr_entry->status != DPM_QUEUED) {	/* No need to process */
			dpm_abort_tr (dbfd);
			if (cpr_entry->status == DPM_ABORTED ||
			    (cpr_entry->status & 0xF000) == DPM_FAILED)
				nb_file_err++;
			continue;
		}
		if ((p = strdup (cpr_entry->from_surl)) == NULL ||
		    (q = strdup (cpr_entry->to_surl)) == NULL) {
			cpr_entry->status = DPM_FAILED | ENOMEM;
			(void) upd_cpr (dbfd, &rec_addrf, cpr_entry);
			continue;
		}
		cpr_entry->status = DPM_ACTIVE;
		(void) upd_cpr (dbfd, &rec_addrf, cpr_entry);
		(getfilereqs+j)->from_surl = p;
		(putfilereqs+j)->to_surl = q;
		(putfilereqs+j)->f_lifetime = cpr_entry->f_lifetime;
		(putfilereqs+j)->f_type =
		    cpr_entry->f_type == '_' ?  '\0' : cpr_entry->f_type;
		strcpy ((putfilereqs+j)->s_token, cpr_entry->s_token);
		(putfilereqs+j)->ret_policy =
		    cpr_entry->ret_policy == '_' ?  '\0' : cpr_entry->ret_policy;
		(putfilereqs+j)->ac_latency =
		    cpr_entry->ac_latency == '_' ?  '\0' : cpr_entry->ac_latency;
		j++;
	}

	if (j == 0)
		return (0);	/* nothing to be done */

	dpm_seterrbuf (errbuf, sizeof(errbuf));

	if (get_srm_endpoint (putfilereqs->to_surl, srm_endpoint, sizeof(srm_endpoint)) < 0) {
		dpm_req->status = DPM_FAILED | EINVAL;
		strcpy (dpm_req->errstring, "SRM endpoint too long");
		return (-1);
	}

	/* get source TURL(s) */

	dpmlogit (func, "calling dpm_get\n");
	errbuf[0] = '\0';
	*r_tokeng = '\0';
	statusg = dpm_get (j, getfilereqs, 1, protocols, NULL, dpm_req->retrytime,
	    r_tokeng, &nbgetreplies, &getfilestatuses);

	if (*r_tokeng)
		dpmlogit (func, "dpm_get returned request token %s\n", r_tokeng);

	/* wait for any source TURL ready */

	while (1) {
		for (i = 0; i < j; i++) {
			if (*(filstatus+i))
				continue;	/* already processed */
			if (! getfilestatuses) {
				if (statusg < 0) {
					*(filstatus+i) = DPM_FAILED | serrno;
					dpmcopy_set_errstring ("get failed", errbuf, NULL,
					    errstring, sizeof(errstring));
					(void) upd_cpr2 (dbfd, dpm_req->r_token,
					    (getfilereqs+i)->from_surl,
					    (putfilereqs+i)->to_surl,
					    0, *(filstatus+i),
					    errstring, 0, cpr_entry);
				}
				continue;
			}
			if (statusg >= 0 && ((getfilestatuses+i)->status == DPM_QUEUED ||
			    (getfilestatuses+i)->status == DPM_ACTIVE))
				continue;
			if ((getfilestatuses+i)->status != DPM_READY) {
				if ((getfilestatuses+i)->status < DPM_READY)
					*(filstatus+i) = statusg < 0 ? (DPM_FAILED | serrno) : statusg;
				else
					*(filstatus+i) = (getfilestatuses+i)->status;
				dpmcopy_set_errstring ("get failed", errbuf,
				    (getfilestatuses+i)->errstring,
				    errstring, sizeof(errstring));
				(void) upd_cpr2 (dbfd, dpm_req->r_token,
				    (getfilereqs+i)->from_surl, (putfilereqs+i)->to_surl,
				    0, *(filstatus+i), errstring, 0, cpr_entry);
				continue;
			}

			/* source TURL ready */

			(putfilereqs+i)->requested_size = (getfilestatuses+i)->filesize;

			/* get destination TURL */

			errbuf[0] = '\0';
			r1 = 0;
			putfilestatus = NULL;
			r_tokenp = NULL;
			dpmlogit (func, "calling srmv2_put\n");
			status = srmv2_put (1, putfilereqs+i, 1, protocols,
			    (dpm_req->flags & 1), dpm_req->retrytime, &r_tokenp,
			    &nbputreplies, &putfilestatus);

			if (r_tokenp)
				dpmlogit (func, "srmv2_put returned request token %s\n", r_tokenp);

			/* wait for destination TURL ready */

			while (status == DPM_QUEUED || status == DPM_ACTIVE) {
				dpm_free_pfilest (nbputreplies, putfilestatus);
				sleep ((r1++ == 0) ? 1 : DEFPOLLINT);
				errbuf[0] = '\0';
				putfilestatus = NULL;
				status = srmv2_getstatus_putreq (r_tokenp, 0, NULL,
				    srm_endpoint, &nbputreplies, &putfilestatus,
				    dpm_req->retrytime);
			}
			if (! putfilestatus) {
				*(filstatus+i) = status < 0 ? (DPM_FAILED | serrno) : status;
				dpmcopy_set_errstring ("put failed", errbuf, NULL,
				    errstring, sizeof(errstring));
				goto upd_cfr;
			}
			if (putfilestatus->status != DPM_READY) {
				if (putfilestatus->status < DPM_FAILED)
					*(filstatus+i) = status < 0 ? (DPM_FAILED | serrno) : status;
				else
					*(filstatus+i) = putfilestatus->status;
				dpmcopy_set_errstring ("put failed", errbuf,
				    putfilestatus->errstring,
				    errstring, sizeof(errstring));
				goto upd_cfr;
			}

			/* destination TURL ready, copy file */

			errbuf[0] = '\0';
			*(filstatus+i) = DPM_ACTIVE;
			dpmlogit (func, "copying %s to %s\n",
			    (getfilestatuses+i)->turl, putfilestatus->turl);
			status = dpm_copyfile ((getfilestatuses+i)->turl,
			    putfilestatus->turl, 5, 0);
			filestatuses = NULL;
			if (status < 0) {
				dpmlogit (func, "dpm_copyfile returned %d, serrno = %d\n",
				    status, serrno);
				status = DPM_FAILED | serrno;
				dpmcopy_set_errstring ("copy failed", errbuf, NULL,
				    errstring, sizeof(errstring));
				(void) srmv2_abortfiles (r_tokenp, 1, &putfilestatus->to_surl,
				    &nbreplies, &filestatuses, dpm_req->retrytime);
			} else {
				dpmlogit (func, "dpm_copyfile returned %d\n", status);
				errbuf[0] = '\0';
				dpmlogit (func, "calling srmv2_putdone\n");
				status = srmv2_putdone (r_tokenp, 1, &putfilestatus->to_surl,
				    &nbreplies, &filestatuses, dpm_req->retrytime);
				if (status < 0) {
					status = DPM_FAILED | serrno;
					dpmcopy_set_errstring ("putdone failed", errbuf, NULL,
					    errstring, sizeof(errstring));
				} else
					errstring[0] = '\0';
			}
			dpm_free_filest (nbreplies, filestatuses);
			*(filstatus+i) = status ? status : DPM_DONE;

			/* try to get target SURL lifetime */

			if (status == 0) {
				dpm_free_pfilest (nbputreplies, putfilestatus);
				putfilestatus = NULL;
				if (srmv2_getstatus_putreq (r_tokenp, 0, NULL,
				    srm_endpoint, &nbputreplies, &putfilestatus,
				    dpm_req->retrytime) == 0 &&
				    putfilestatus)
					cpr_entry->f_lifetime = putfilestatus->f_lifetime;
			}
upd_cfr:
			filestatuses = NULL;
			(void) dpm_relfiles (r_tokeng, 1, &(getfilestatuses+i)->from_surl,
			    0, &nbreplies, &filestatuses);
			dpm_free_filest (nbreplies, filestatuses);
			(void) upd_cpr2 (dbfd, dpm_req->r_token,
			    (getfilereqs+i)->from_surl, (putfilereqs+i)->to_surl,
			    (*(filstatus+i) == DPM_DONE) ? (getfilestatuses+i)->filesize : 0,
			    (*(filstatus+i) == DPM_DONE) ? 0 : *(filstatus+i),
			    errstring, cpr_entry->f_lifetime, cpr_entry);
			dpm_free_pfilest (nbputreplies, putfilestatus);
			if (r_tokenp)
				free (r_tokenp);
		}

		for (i = 0; i < j; i++) {
			if (*(filstatus+i) == 0 || *(filstatus+i) == DPM_ACTIVE)
				break;
		}
		if (i >= j) break;	/* no more pending file request */

		dpm_free_gfilest (nbgetreplies, getfilestatuses);
		getfilestatuses = NULL;
		errbuf[0] = '\0';
		sleep ((r++ == 0) ? 1 : DEFPOLLINT);
		statusg = dpm_getstatus_getreq (r_tokeng, 0, NULL,
		    &nbgetreplies, &getfilestatuses);
	}
	return (0);
}