/*

 * lftp - file transfer program

 *

 * Copyright (c) 1996-2017 by Alexander V. Lukyanov (lav@yars.free.net)

 *

 * This program is free software; you can redistribute it and/or modify

 * it under the terms of the GNU General Public License as published by

 * the Free Software Foundation; either version 3 of the License, or

 * (at your option) any later version.

 *

 * This program is distributed in the hope that it will be useful,

 * but WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

 * GNU General Public License for more details.

 *

 * You should have received a copy of the GNU General Public License

 * along with this program.  If not, see <http://www.gnu.org/licenses/>.

 */

/* FileCopyPeer behaviour:

    1) when suspended, does nothing

    2) tries to read some data at seek_pos, sets pos to position of Get (get).

    2.5) tries to position to seek_pos and gets ready to write (put).

    3) if it cannot seek to seek_pos, changes pos to what it can seek.

    4) if it knows that it cannot seek to pos>0, CanSeek()==false

    5) if it knows that it cannot seek to pos==0, CanSeek0()==false

    6) it tries to get date/size if told to. (get)

    7) it sets date on the file if eof is reached and date is known (put).

    8) if put needs size/date before it writes data, NeedSizeDateBeforehand()==true.

 */

#include <config.h>

#include <assert.h>

#include <unistd.h>

#include <errno.h>

#include <sys/types.h>

#include <sys/stat.h>

#include <fcntl.h>

#include <math.h>

#include <stddef.h>

#include "FileCopy.h"

#include "url.h"

#include "log.h"

#include "misc.h"

#include "LsCache.h"

#include "plural.h"

#include "ArgV.h"

#define skip_threshold 0x1000

ResDecl rate_period  ("xfer:rate-period","15", ResMgr::UNumberValidate,ResMgr::NoClosure);

ResDecl eta_period   ("xfer:eta-period", "120",ResMgr::UNumberValidate,ResMgr::NoClosure);

ResDecl max_redir    ("xfer:max-redirections", "5",ResMgr::UNumberValidate,ResMgr::NoClosure);

ResDecl buffer_size  ("xfer:buffer-size","0x10000",ResMgr::UNumberValidate,ResMgr::NoClosure);

// It's bad when lftp receives data in small chunks, try to accumulate

// data in a kernel buffer using a delay and slurp it at once:

enum {

   // Delays in microseconds

   MAX_DELAY=30000,

   DELAY_STEP=30,

   // This size is related to socket buffer size.

   // When it is too large, tcp slowdown happens.

   // SSL has packet size 0x4000, so we have to use a lower threshold.

   MAX_READ_TO_DELAY=0x3F00,

};

// FileCopy

#define super SMTask

#define set_state(s) do { state=(s); \

   Log::global->Format(11,"FileCopy(%p) enters state %s\n", this, #s); } while(0)

int FileCopy::Do()

{

   int m=STALL;

   const char *b;

   int s;

   int rate_add;

   if(Error() || Done())

      return m;

   switch(state)

   {

   pre_INITIAL:

      set_state(INITIAL);

      m=MOVED;

   case(INITIAL):

      if(remove_target_first && !put->FileRemoved())

	 return m;

      remove_target_first=false;

      if(cont && put->CanSeek())

	 put->WantSize();

      if(put->NeedSizeDateBeforehand() || (cont && put->CanSeek() && put->GetSize()==NO_SIZE_YET))

      {

	 if(get->GetSize()==NO_SIZE_YET || get->GetDate()==NO_DATE_YET)

	 {

	    put->Suspend();

	    get->DontStartTransferYet();

	    get->Resume();

	    get->WantSize();

	    if(put->NeedDate())

	       get->WantDate();

	    goto pre_GET_INFO_WAIT;

	 }

      }

      if(get->GetSize()==NO_SIZE_YET)

	 get->WantSize();

      if(get->GetSize()!=NO_SIZE && get->GetSize()!=NO_SIZE_YET)

	 put->SetEntitySize(get->GetSize());

      if(get->GetDate()!=NO_DATE && get->GetDate()!=NO_DATE_YET)

	 put->SetDate(get->GetDate());

      else if(get->GetDate()==NO_DATE_YET)

      {

	 if(put->NeedDate())

	    get->WantDate();

      }

      if(cont && put->CanSeek())

	 put->Seek(FILE_END);

      else

      {

	 if(put->range_start>0 && put->CanSeek())

	    put->Seek(put->range_start);

	 if(get->range_start>0 && get->CanSeek())

	    get->Seek(get->range_start);

	 goto pre_DO_COPY;

      }

      get->Suspend();

      put->Resume();

      set_state(PUT_WAIT);

      m=MOVED;

      /* fallthrough */

   case(PUT_WAIT):

      if(put->Error())

	 goto put_error;

      if(put->GetSeekPos()!=FILE_END && get->GetSize()>=0

      && put->GetSeekPos()>=get->GetSize())

      {

	 debug((9,_("copy: destination file is already complete\n")));

	 if(get->GetDate()!=NO_DATE)

	    goto pre_CONFIRM_WAIT;  // have to set the date.

	 goto pre_GET_DONE_WAIT;

      }

      if(!put->IOReady())

	 return m;

      /* now we know if put's seek failed. Seek get accordingly. */

      if(get->CanSeek())

	 get->Seek(put->GetRealPos());

   pre_DO_COPY:

      get->Resume();

      get->StartTransfer();

      RateReset();

      set_state(DO_COPY);

      m=MOVED;

      /* fallthrough */

   case(DO_COPY): {

      if(put->Error())

      {

      put_error:

	 SetError(put->ErrorText());

	 return MOVED;

      }

      if(get->Error() && get->Size()==0)

      {

	 put->DontVerify();

	 if(put->GetPos()>0)

	 {

	    put->PutEOF();

	    put->Roll();

	 }

      get_error:

	 SetError(get->ErrorText());

	 return MOVED;

      }

      if(put->Broken())

      {

	 get->Suspend();

	 if(!put->Done())

	    return m;

	 debug((9,_("copy: put is broken\n")));

	 if(fail_if_broken)

	 {

	    SetError(strerror(EPIPE));

	    return MOVED;

	 }

	 goto pre_GET_DONE_WAIT;

      }

      put->Resume();

      if(put->GetSeekPos()==FILE_END)   // put position is not known yet.

      {

	 get->Suspend();

	 return m;

      }

      get->Resume();

      if(fail_if_cannot_seek && (get->GetRealPos()<get->range_start

			      || put->GetRealPos()<put->range_start

			      || get->GetRealPos()!=put->GetRealPos()))

      {

	 SetError(_("seek failed"));

	 return MOVED;

      }

      if(high_watermark_timeout.Stopped())

      {

	 SetError(_("no progress timeout"));

	 return MOVED;

      }

      if(get->GetSize()>0 && get->GetRealPos()>get->GetSize())

      {

	 get->SetSize(NO_SIZE_YET);

	 get->SetDate(NO_DATE_YET);

      }

      long lbsize=0;

      if(line_buffer)

	 lbsize=line_buffer->Size();

      /* check if positions are correct */

      off_t get_pos=get->GetRealPos()-get->range_start;

      off_t put_pos=put->GetRealPos()-put->range_start;

      if(get_pos-lbsize!=put_pos)

      {

	 if(line_buffer)

	    line_buffer->Empty();

	 if(get_pos==put_pos)

	 {  // rare case.

	    return MOVED;

	 }

	 if(put_pos

	 {

	    if(!get->CanSeek(put->GetRealPos()))

	    {

	       // we lose... How about a large buffer?

	       SetError(_("cannot seek on data source"));

	       return MOVED;

	    }

	    debug((9,_("copy: put rolled back to %lld, seeking get accordingly\n"),

		     (long long)put->GetRealPos()));

	    debug((10,"copy: get position was %lld\n",

		     (long long)get->GetRealPos()));

	    get->Seek(put->GetRealPos());

	    return MOVED;

	 }

	 else // put_pos > get_pos

	 {

	    off_t size=get->GetSize();

	    if(size>=0 && put->GetRealPos()>=size)

	    {

	       // simulate eof, as we have already have the whole file.

	       debug((9,_("copy: all data received, but get rolled back\n")));

	       goto eof;

	    }

	    off_t skip=put->GetRealPos()-get->GetRealPos();

	    if(!put->CanSeek(get->GetRealPos()) || skip

	    {

	       // we have to skip some data

	       get->Get(&b,&s);

	       if(skip>s)

		  skip=s;

	       if(skip==0)

		  return m;

	       get->Skip(skip);

	       bytes_count+=skip;

	       return MOVED;

	    }

	    debug((9,_("copy: get rolled back to %lld, seeking put accordingly\n"),

		     (long long)get->GetRealPos()));

	    put->Seek(get->GetRealPos());

	    return MOVED;

	 }

      }

      if(put->IsFull())

	 get->Suspend(); // stall the get.

      get->Get(&b,&s);

      if(b==0) // eof

      {

	 debug((10,"copy: get hit eof\n"));

	 goto eof;

      }

      rate_add=put_buf;

      if(s==0)

      {

	 put_buf=put->Buffered();

	 rate_add-=put_buf;

	 RateAdd(rate_add);

	 if(put->Size()==0)

	    put->Suspend();

	 return m;

      }

      m=MOVED;

      if(get->range_limit!=FILE_END && get->range_limit<get->GetRealPos()+s)

      {

	 s=get->range_limit-get->GetRealPos();

	 if(s<0)

	    s=0;

      }

      if(line_buffer)

      {

	 const char *lb;

	 int ls;

	 if(line_buffer->Size()>line_buffer_max)

	 {

	    line_buffer->Get(&lb,&ls);

	    put->Put(lb,ls);

	    line_buffer->Skip(ls);

	 }

	 line_buffer->Put(b,s);

	 get->Skip(s);

	 bytes_count+=s;

	 // now find eol in line_buffer.

	 line_buffer->Get(&lb,&ls);

	 const char *eol=0;

	 if(get->Eof() || get->Error())

	    eol=lb+ls-1;

	 else

	    eol=memrchr(lb,'\n',ls);

	 if(eol)

	 {

	    put->Put(lb,eol-lb+1);

	    line_buffer->Skip(eol-lb+1);

	 }

      }

      else

      {

	 put->Put(b,s);

	 get->Skip(s);

	 bytes_count+=s;

      }

      put_buf=put->Buffered();

      rate_add-=put_buf-s;

      RateAdd(rate_add);

      if(high_watermark

      {

	 high_watermark=put_pos+s;

	 high_watermark_timeout.Reset();

      }

      if(get->range_limit!=FILE_END && get->range_limit<=get->GetRealPos())

      {

	 debug((10,"copy: get reached range limit\n"));

	 goto eof;

      }

      return m;

   }

   eof:

      if(line_buffer)

      {

	 line_buffer->Get(&b,&s);

	 put->Put(b,s);

	 line_buffer->Skip(s);

      }

      if(!CheckFileSizeAtEOF())

      {

	 SetError(_("file size decreased during transfer"));

	 return MOVED;

      }

   pre_CONFIRM_WAIT:

      put->SetSuggestedFileName(get->GetSuggestedFileName());

      put->SetDate(get->GetDate());

      if(get->GetSize()!=NO_SIZE && get->GetSize()!=NO_SIZE_YET)

	 put->SetEntitySize(get->GetSize());

      put->PutEOF();

      get->Suspend();

      put->Resume();

      put_eof_pos=put->GetRealPos();

      debug((10,"copy: waiting for put confirmation\n"));

      set_state(CONFIRM_WAIT);

      m=MOVED;

   case(CONFIRM_WAIT):

      if(put->Error())

	 goto put_error;

      /* check if put position is correct */

      if(put_eof_pos!=put->GetRealPos() || put->GetSeekPos()==FILE_END)

      {

	 set_state(DO_COPY);

	 return MOVED;

      }

      rate_add=put_buf;

      put_buf=put->Buffered();

      rate_add-=put_buf;

      RateAdd(rate_add);

      if(!put->Done())

	 return m;

      debug((10,"copy: put confirmed store\n"));

   pre_GET_DONE_WAIT:

      get->Empty();

      get->PutEOF();

      get->Resume();

      set_state(GET_DONE_WAIT);

      m=MOVED;

      end_time=now;

      put->Suspend();

      /* fallthrough */

   case(GET_DONE_WAIT):

      if(get->Error())

	 goto get_error;

      if(remove_source_later)

      {

	 get->RemoveFile();

	 remove_source_later=false;

      }

      if(!get->Done())

	 return m;

      debug((10,"copy: get is finished - all done\n"));

      set_state(ALL_DONE);

      get->Suspend();

      LogTransfer();

      return MOVED;

   pre_GET_INFO_WAIT:

      set_state(GET_INFO_WAIT);

      m=MOVED;

   case(GET_INFO_WAIT):

      if(get->Error())

	 goto get_error;

      if(get->GetSize()==NO_SIZE_YET || get->GetDate()==NO_DATE_YET)

	 return m;

      goto pre_INITIAL;

   case(ALL_DONE):

      return m;

   }

   return m;

}

FileCopy::FileCopy(FileCopyPeer *s,FileCopyPeer *d,bool c)

   : get(s), put(d), cont(c),

   rate("xfer:rate-period"),

   rate_for_eta("xfer:eta-period"),

   high_watermark_timeout("xfer:timeout",0)

{

   set_state(INITIAL);

   int max_buf=buffer_size.Query(0);

   if(max_buf<1)

      max_buf=1;

   s->SetMaxBuffered(max_buf);

   d->SetMaxBuffered(max_buf);

   put_buf=0;

   put_eof_pos=0;

   high_watermark=0;

   bytes_count=0;

   fail_if_cannot_seek=false;

   fail_if_broken=true;

   remove_source_later=false;

   remove_target_first=false;

   line_buffer_max=0;

}

FileCopy::~FileCopy()

{

}

FileCopy *FileCopy::New(FileCopyPeer *s,FileCopyPeer *d,bool c)

{

   FileCopy *res=0;

   if(fxp_create)

      res=fxp_create(s,d,c);

   if(res)

      return res;

   return new FileCopy(s,d,c);

}

void FileCopy::SuspendInternal()

{

   super::SuspendInternal();

   if(get) get->SuspendSlave();

   if(put) put->SuspendSlave();

}

void FileCopy::ResumeInternal()

{

   if(get) get->ResumeSlave();

   if(put) put->ResumeSlave();

   super::ResumeInternal();

}

void FileCopy::Fg()

{

   if(get) get->Fg();

   if(put) put->Fg();

}

void FileCopy::Bg()

{

   if(get) get->Bg();

   if(put) put->Bg();

}

void FileCopy::SetError(const char *str)

{

   error_text.set(str);

   get=0;

}

void FileCopy::LineBuffered(int s)

{

   if(!line_buffer)

      line_buffer=new Buffer();

   line_buffer_max=s;

}

off_t FileCopy::GetPos() const

{

   if(put) {

      off_t pos = put->GetRealPos() - put->Buffered();

      // sometimes Buffered overestimates the amount of buffered data

      if(pos<0)

	 pos=0;

      return pos;

   }

   if(get)

      return get->GetRealPos();

   return 0;

}

off_t FileCopy::GetSize() const

{

   if(get)

      return get->GetSize();

   return NO_SIZE;

}

int FileCopy::GetPercentDone() const

{

   if(!get || !put)

      return 100;

   off_t size=get->GetSize();

   if(size==NO_SIZE || size==NO_SIZE_YET)

      return -1;

   if(size==0)

      return 0;

   off_t ppos=put->GetRealPos() - put->Buffered() - put->range_start;

   if(ppos<0)

      return 0;

   off_t psize=size-put->range_start;

   if(put->range_limit!=FILE_END)

      psize=put->range_limit-put->range_start;

   if(psize<0)

      return 100;

   if(ppos>psize)

      return -1;

   return percent(ppos,psize);

}

const char *FileCopy::GetPercentDoneStr() const

{

   int pct=GetPercentDone();

   if(pct==-1)

      return "";

   static char buf[8];

   snprintf(buf,8,"(%d%%) ",pct);

   return buf;

}

void FileCopy::RateAdd(int a)

{

   rate.Add(a);

   rate_for_eta.Add(a);

}

void FileCopy::RateReset()

{

   start_time=now;

   rate.Reset();

   rate_for_eta.Reset();

}

float FileCopy::GetRate()

{

   if(!rate.Valid() || !put)

      return 0;

   return rate.Get();

}

const char *FileCopy::GetRateStr()

{

   if(!rate.Valid() || !put)

      return "";

   return rate.GetStrS();

}

off_t FileCopy::GetBytesRemaining()

{

   if(!get)

      return 0;

   if(get->range_limit==FILE_END)

   {

      off_t size=get->GetSize();

      if(size<=0 || size<get->GetRealPos() || !rate_for_eta.Valid())

	 return -1;

      return(size-GetPos());

   }

   return get->range_limit-GetPos();

}

const char *FileCopy::GetETAStr()

{

   off_t b=GetBytesRemaining();

   if(b<0 || !put)

      return "";

   return rate_for_eta.GetETAStrSFromSize(b);

}

long FileCopy::GetETA(off_t b)

{

   if(b<0 || !rate_for_eta.Valid())

      return -1;

   return (long)(double(b) / rate_for_eta.Get() + 0.5);

}

const char *FileCopy::GetStatus()

{

   static xstring buf;

   const char *get_st=get?get->GetStatus():0;

   const char *put_st=put?put->GetStatus():0;

   if(get_st && get_st[0] && put_st && put_st[0])

      buf.vset("[",get_st,"->",put_st,"]",NULL);

   else if(get_st && get_st[0])

      buf.vset("[",get_st,"]",NULL);

   else if(put_st && put_st[0])

      buf.vset("[",put_st,"]",NULL);

   else

      return "";

   return buf;

}

double FileCopy::GetTimeSpent()

{

   if(end_time

      return 0;

   return TimeDiff(end_time,start_time);

}

FgData *FileCopy::GetFgData(bool fg)

{

   // NOTE: only one of get/put can have FgData in this implementation.

   FgData *f=0;

   if(get) f=get->GetFgData(fg);

   if(f) return f;

   if(put) f=put->GetFgData(fg);

   return f;

}

pid_t FileCopy::GetProcGroup()

{

   pid_t p=0;

   if(get) p=get->GetProcGroup();

   if(p) return p;

   if(put) p=put->GetProcGroup();

   return p;

}

void FileCopy::Kill(int sig)

{

   if(get) get->Kill(sig);

   if(put) put->Kill(sig);

}

Ref<Log> FileCopy::transfer_log;

void FileCopy::LogTransfer()

{

   const char *log_ctx="xfer";

   if(!ResMgr::QueryBool("log:enabled",log_ctx))

      return;

   const char *src=get->GetURL();

   if(!src)

      return;

   src=alloca_strdup(src);

   const char *dst=put->GetURL();

   if(!dst)

      return;

   dst=alloca_strdup(dst);

   if(!transfer_log)

      transfer_log=new Log(log_ctx);

   long long range_limit=GetRangeLimit();

   if(range_limit==FILE_END)

      range_limit=get->GetPos();

   transfer_log->Format(0,"%s -> %s %lld-%lld %s\n",

      url::remove_password(src),url::remove_password(dst),

      (long long)GetRangeStart(),range_limit,

      Speedometer::GetStrProper(GetBytesCount()/GetTimeSpent()).get());

}

void FileCopy::SetRange(off_t s,off_t lim)

{

   get->SetRange(s,lim);

   put->SetRange(s,lim);

}

bool FileCopy::CheckFileSizeAtEOF() const

{

   long long range_limit=GetRangeLimit();

   if(range_limit==FILE_END)

   {

      const long long size=GetSize();

      if(size==NO_SIZE || size==NO_SIZE_YET)

	 return true;   // nothing to compare with.

      range_limit=size;

   }

   const long long get_pos=get->GetRealPos();

   const long long put_pos=put->GetRealPos();

   const long long pos=(get_pos>put_pos ? get_pos : put_pos);

   if(pos<=0 || pos>=range_limit)

      return true;

   debug((0,"expected pos=%lld, actual pos=%lld\n",range_limit,pos));

   return false;

}

// FileCopyPeer implementation

#undef super

#define super Buffer

off_t FileCopyPeer::GetSize()

{

   if(size>=0 && pos>size)

      WantSize();

   return size;

}

void FileCopyPeer::SetSize(off_t s)

{

   size=s;

   if(seek_pos==FILE_END)

   {

      if(size!=NO_SIZE && size!=NO_SIZE_YET)

	 seek_pos=size;

      else

	 seek_pos=0;

   }

}

void FileCopyPeer::SetDate(time_t d,int p)

{

   date.set(d,p);

   if(d==NO_DATE || d==NO_DATE_YET)

      date_set=true;

   else

      date_set=false;

}

void FileCopyPeer::SetRange(const off_t s,const off_t lim)

{

   range_start=s;

   range_limit=lim;

   if(mode==PUT || range_start>GetPos()+0x4000)

      Seek(range_start);

}

bool FileCopyPeer::Done()

{

   if(Error())

      return true;

   if(eof && Size()==0)

   {

      if(removing)

	 return false;

      if(mode==PUT)

	 return done;

      return true;

   }

   if(broken)

      return true;

   return false;

}

void FileCopyPeer::Seek(off_t offs)

{

   seek_pos=offs;

   if(mode==PUT)

      pos-=Size();

   Empty();

   eof=false;

   broken=false;

}

const char *FileCopy::TempFileName(const char *file)

{

   if(!ResMgr::QueryBool("xfer:use-temp-file",0))

      return file;

   xstring &temp=xstring::get_tmp(ResMgr::Query("xfer:temp-file-name",0));

   if(temp.length()==0 || temp.eq("*"))

      return file;

   const char *name=basename_ptr(file);

   int subst_pos=temp.instr('*');

   if(subst_pos>=0)

      temp.set_substr(subst_pos,1,name);

   else {

      if(temp.last_char()=='.')

	 temp.append(name);

      else if(temp[0]=='.')

	 temp.set_substr(0,0,name);

      else

	 temp.append('.').append(name);

   }

   return dir_file(dirname(file),temp);

}

const char *FileCopyPeer::UseTempFile(const char *file)

{

   const char *temp=FileCopy::TempFileName(file);

   if(temp==file)

      return file;

   auto_rename=true;

   temp_file=true;

   SetSuggestedFileName(basename_ptr(file));

   return temp;

}

FileCopyPeer::FileCopyPeer(dir_t m) : IOBuffer(m)

{

   want_size=false;

   want_date=false;

   start_transfer=true;

   size=NO_SIZE_YET;

   date=NO_DATE_YET;

   e_size=NO_SIZE;

   seek_pos=0;

   can_seek=false;

   can_seek0=false;

   date_set=false;

   do_set_date=true;

   do_verify=true;

   ascii=false;

   range_start=0;

   range_limit=FILE_END;

   removing=false;

   file_removed=false;

   temp_file=false;

   do_mkdir=false;

   use_cache=true;

   write_allowed=true;

   done=false;

   auto_rename=false;

   Suspend();  // don't do anything too early

}

// FileCopyPeerFA implementation

#undef super

#define super FileCopyPeer

int FileCopyPeerFA::Do()

{

   int m=STALL;

   int res;

   if(session->OpenMode()==FA::MAKE_DIR)

   {

      // doing mkdir

      int res=session->Done();

      if(res==FA::IN_PROGRESS)

	 return m;

      if(res<0)

	 debug((3,"mkdir failed: %s\n",session->StrError(res)));

      session->Close();

      m=MOVED;

   }

   else if(session->OpenMode()==FA::RENAME)

   {

      int res=session->Done();

      if(res==FA::IN_PROGRESS)

	 return m;

      if(res<0) {

	 if(temp_file)

	    SetError(session->StrError(res));

	 else

	    debug((3,"rename failed: %s\n",session->StrError(res)));

      }

      session->Close();

      done=true;

      return MOVED;

   }

   if(do_mkdir) {

      // do mkdir just once

      do_mkdir=false;

      assert(!session->IsOpen());

      const xstring& dir=dirname(file);

      if(dir.length()>0 && dir.ne("/") && dir.ne(".") && dir.ne("..")) {

	 // FIXME: .././.. should be also excluded