#!/usr/bin/perl

## $Id: obec.fetcher.pl,v 1.34 2003/04/09 06:37:50 lord Exp $
##-----------------------------------------------------
## NNTP->RDBMS gateway for OBEC project.
## http://obec.sourceforge.net/
## Vadim Zaliva <lord@crocodile.org>
##-----------------------------------------------------
## 
## TODO:
## 1. Error handling
## 2. Better MIME body extraction - find first text/plain part, etc.
## 3. Try to guess broken encodings, like koi8 converted via cp1251->koi8 filter.
## 4. Use fetch rows as arrays, opposed to hashrefs.
##
## BUGS:
##
##
##-----------------------------------------------------

use Net::NNTP             ;
use ConfigReader::Simple  ;
use DBI                   ;
use Mail::Address         ;
use Lingua::DetectCharset ;
use Convert::Cyrillic     ;
use MIME::Head            ;
use MIME::Parser          ;
use Date::Parse           ;
use POSIX qw(strftime)    ;

# ---------------- Constants --------------------------

$DBCHARSET = 'KOI8';

# ---------------------- Main -------------------------

# Load CFG.
$config    = load_cfg("obec.fetcher.cfg");
@articles_blacklist = load_blacklist("articles.blacklist.cfg");
@comments_blacklist = load_blacklist("comments.blacklist.cfg");

# Connect to DB
($dbh, $drh) = db_connect($config);

# Connect to NNTP server
$nntp = nntp_connect($config);

# Do actuall work
process_articles_group($nntp, $config->get("ArticlesGroup"));
process_comments_group($nntp, $config->get("CommentsGroup"));

# Close NNTP connection
$nntp->quit;

find_threads($config->get("UnsolvedKeepDays"));

# Disconnect from the database.
db_disconnect();

print "Done.\n";

exit(0);

# ---------------------- Subroutines -------------------------

sub trim($)
  {
    my @out = @_;
    for(@out) {
      s/^\s+//;
      s/\s+$//;
    }
    return wantarray ? @out:$out[0];
  }
  
sub load_cfg($)
  {
    my $config = ConfigReader::Simple->new($_[0], [qw(Server DSN DBUser DBPassword ArticlesGroup)]);
    $config->parse();
    $config->_validate_keys();  
    return $config;
  };
             
sub load_blacklist($)
{
  print "Reading blacklist from @_[0]\n";
  my @res;
  my $i=0;
  open(BF,"< $_[0]") or return -1;
  while(<BF>)
    {
      $l=$_;
      chop $l;
      if(!(($l =~ /^\#/) || length($l)==0))
        {
          #print "Adding '$l' to blacklist.\n";
          @res[$i++]=$l;
        }
    }
  
  close BF;
  return @res;
};
             
sub db_disconnect()
  {
    $dbh->disconnect();
  }

sub db_connect($)
{
  my($config)=@_;

  my $dsn         = $config->get("DSN");
  my $db_user     = $config->get("DBUser");
  my $db_password = $config->get("DBPassword");
  
  print "Connecting to: $dsn as $db_user\n";
  $dbh = DBI->connect($dsn, $db_user, $db_password);
  $drh = DBI->install_driver("mysql");

  $select_email_id_by_msgid    = $dbh->prepare("SELECT Id FROM EmailMessages WHERE MsgId=?");
  $select_author_id_by_email   = $dbh->prepare("SELECT Id FROM Authors WHERE Email=?");
  $insert_author               = $dbh->prepare("INSERT INTO Authors (Name, Email) VALUES (?, ?)");
  $insert_email                = $dbh->prepare("INSERT INTO EmailMessages (MsgId, FTNMsgId, Header) VALUES (?, ?, ?)");
  $insert_text                 = $dbh->prepare("INSERT INTO Texts (AuthorId, PublishedAt, EmailId, Title, Bytes, Text, Views) VALUES (?, ?, ?, ?, ?, ?, ?)");

  $set_new_texts_thread_by_text_id        = $dbh->prepare("UPDATE Texts SET CommentThreadId=? WHERE Id=?");

  $insert_comment              = $dbh->prepare("INSERT INTO Comments (AuthorId, PublishedAt, EmailId, Title, Bytes, Text, ThreadId, ThreadPosition, ThreadOffset) VALUES (?, ?, ?, ?, ?, ?, ?, 0, 0)");

  $shuffle_thread              = $dbh->prepare("UPDATE Comments SET ThreadPosition=ThreadPosition+? WHERE ThreadId = ? AND ThreadPosition >= ?");

  $start_thread                = $dbh->prepare("INSERT INTO Threads (CreatedAt, Title) VALUES (NOW(), ?)");

  $insert_unsolved_msgid       = $dbh->prepare("INSERT INTO UnsolvedMsgIdReferences (ThreadId, MsgId) VALUES (?,?)");

  $insert_unsolved_ftnmsgid    = $dbh->prepare("INSERT INTO UnsolvedFTNMsgIdReferences (ThreadId, FTNMsgId) VALUES (?,?)");

  $comments_to_comments_by_ftnmsgid = $dbh->prepare("select Comments.Id, Comments.ThreadId, Comments.ThreadPosition, Comments.ThreadOffset, UnsolvedFTNMsgIdReferences.ThreadId AS CThreadId from Comments, EmailMessages, UnsolvedFTNMsgIdReferences WHERE EmailMessages.Id=Comments.EmailId AND UnsolvedFTNMsgIdReferences.FTNMsgId=EmailMessages.FTNMsgId ORDER BY Comments.PublishedAt LIMIT 1");

  $remove_unsolved_ftnmsgid    = $dbh->prepare("DELETE FROM UnsolvedFTNMsgIdReferences WHERE ThreadId=?");

  $link_thread_to_comment     = $dbh->prepare("UPDATE Comments SET ThreadId=?, ThreadOffset=ThreadOffset+?, ThreadPosition=ThreadPosition+? WHERE ThreadId=?");

  $comments_to_comments_by_msgid = $dbh->prepare("select Comments.Id, Comments.ThreadId, Comments.ThreadPosition, Comments.ThreadOffset, UnsolvedMsgIdReferences.ThreadId AS CThreadId from Comments, EmailMessages, UnsolvedMsgIdReferences WHERE EmailMessages.Id=Comments.EmailId AND UnsolvedMsgIdReferences.MsgId=EmailMessages.MsgId ORDER BY Comments.PublishedAt LIMIT 1");

  $remove_unsolved_msgid      = $dbh->prepare("DELETE FROM UnsolvedMsgIdReferences WHERE ThreadId=?");
  
  $comments_to_texts_by_ftnmsgid = $dbh->prepare("SELECT Texts.Id, UnsolvedFTNMsgIdReferences.ThreadId from Texts, EmailMessages, UnsolvedFTNMsgIdReferences, Threads WHERE EmailMessages.Id=Texts.EmailId AND UnsolvedFTNMsgIdReferences.FTNMsgId=EmailMessages.FTNMsgId AND Texts.CommentThreadId IS NULL AND Threads.Id=UnsolvedFTNMsgIdReferences.ThreadId ORDER BY Threads.CreatedAt LIMIT 1");
  
  $comments_to_texts_by_msgid = $dbh->prepare("SELECT Texts.Id, UnsolvedMsgIdReferences.ThreadId from Texts, EmailMessages, UnsolvedMsgIdReferences, Threads WHERE EmailMessages.Id=Texts.EmailId AND UnsolvedMsgIdReferences.MsgId=EmailMessages.MsgId AND Texts.CommentThreadId IS NULL AND Threads.Id=UnsolvedMsgIdReferences.ThreadId ORDER BY Threads.CreatedAt LIMIT 1");

  $comments_to_texts_add_by_ftnmsgid = $dbh->prepare("SELECT Texts.Id, Texts.CommentThreadId, UnsolvedFTNMsgIdReferences.ThreadId from Texts, EmailMessages, UnsolvedFTNMsgIdReferences, Threads WHERE EmailMessages.Id=Texts.EmailId AND UnsolvedFTNMsgIdReferences.FTNMsgId=EmailMessages.FTNMsgId AND Texts.CommentThreadId IS NOT NULL AND Texts.CommentThreadId != UnsolvedFTNMsgIdReferences.ThreadId AND Threads.Id=UnsolvedFTNMsgIdReferences.ThreadId ORDER BY Threads.CreatedAt LIMIT 1");
  
  $comments_to_texts_add_by_msgid = $dbh->prepare("SELECT Texts.Id, Texts.CommentThreadId, UnsolvedMsgIdReferences.ThreadId from Texts, EmailMessages, UnsolvedMsgIdReferences, Threads WHERE EmailMessages.Id=Texts.EmailId AND UnsolvedMsgIdReferences.MsgId=EmailMessages.MsgId AND Texts.CommentThreadId IS NOT NULL AND Texts.CommentThreadId != UnsolvedMsgIdReferences.ThreadId AND Threads.Id=UnsolvedMsgIdReferences.ThreadId ORDER BY Threads.CreatedAt LIMIT 1");

  $remove_empty_threads = $dbh->prepare("SELECT Threads.Id FROM Threads LEFT JOIN Comments ON Comments.ThreadId=Threads.Id GROUP BY Threads.Id HAVING COUNT(Comments.Id)=0");

  $delete_thread        = $dbh->prepare("DELETE FROM Threads WHERE Id=?");

  $count_comments_in_thread  = $dbh->prepare("SELECT COUNT(Id) AS Count FROM Comments WHERE ThreadId=?");

  $unresolved_comments_to_comment_by_ftnmsgid = $dbh->prepare("select A.ThreadId, B.ThreadId AS CThreadId from UnsolvedFTNMsgIdReferences  A, UnsolvedFTNMsgIdReferences B, Threads WHERE A.FTNMsgId=B.FTNMsgId AND A.ThreadId < B.ThreadId AND Threads.Id=B.ThreadId ORDER by A.ThreadId, Threads.CreatedAt LIMIT 1");

  $unresolved_comments_to_comment_by_msgid = $dbh->prepare("select A.ThreadId, B.ThreadId AS CThreadId from UnsolvedMsgIdReferences  A, UnsolvedMsgIdReferences B, Threads WHERE A.MsgId=B.MsgId AND A.ThreadId < B.ThreadId AND Threads.Id=B.ThreadId ORDER by A.ThreadId, Threads.CreatedAt LIMIT 1");

  $change_unresolved_ftn_thread = $dbh->prepare("UPDATE UnsolvedFTNMsgIdReferences SET ThreadId=? WHERE ThreadId=?");
  $change_unresolved_msg_thread = $dbh->prepare("UPDATE UnsolvedMsgIdReferences SET ThreadId=? WHERE ThreadId=?");

  $find_ftn_dupes = $dbh->prepare("select ThreadId, FTNMsgId from UnsolvedFTNMsgIdReferences GROUP BY FTNMsgId HAVING COUNT(*)>1");

  $find_msg_dupes = $dbh->prepare("select ThreadId, MsgId from UnsolvedMsgIdReferences GROUP BY MsgId HAVING COUNT(*)>1");

  $delete_ftn_dupe = $dbh->prepare("delete from UnsolvedFTNMsgIdReferences WHERE ThreadId=? AND FTNMsgId=? LIMIT 1");

  $delete_msg_dupe = $dbh->prepare("delete from UnsolvedMsgIdReferences WHERE ThreadId=? AND MsgId=? LIMIT 1");

  $find_comment_merge_point = $dbh->prepare("select ThreadPosition from Comments, Threads where ThreadPosition>? AND ThreadOffset=?+1 AND PublishedAt>Threads.CreatedAt AND ThreadId=? AND Threads.Id=? ORDER BY PublishedAt ASC LIMIT 1");

  $find_comment_merge_point_next_parent = $dbh->prepare("select MIN(ThreadPosition)+1 AS NewPos from Comments WHERE ThreadPosition > ? AND ThreadOffset <= ? AND ThreadId=?");

  $find_comment_merge_point_last_response = $dbh->prepare("select MAX(ThreadPosition)+1 AS NewPos from Comments where ThreadPosition >= ? AND ThreadId=?");

  $oldest_in_thread = $dbh->prepare("SELECT MIN(PublishedAt) FROM Comments WHERE ThreadId=? AND ThreadOffset=0");

  $set_thread_date = $dbh->prepare("UPDATE Threads SET CreatedAt=? WHERE Id=?");
  $remove_old_ftn_unsolved    = $dbh->prepare("DELETE FROM UnsolvedFTNMsgIdReferences WHERE (TO_DAYS(NOW())-TO_DAYS(CreatedAt))>?");

  $remove_old_msg_unsolved    = $dbh->prepare("DELETE FROM UnsolvedMsgIdReferences WHERE (TO_DAYS(NOW())-TO_DAYS(CreatedAt))>?");

  return ($dbh, $drh);
}

sub nntp_connect($)
{
  my($config)=@_;

  my $server     = $config->get("Server");
  print "Connecting to: $server\n";
  
  $nntp = Net::NNTP->new($server);
  $nntp->reader();

  return $nntp;
}

sub process_comments_group(@)
  {
    my($nntp, $group_name)=@_;
    process_group($nntp, $group_name, \&process_comment);
  }

sub process_articles_group(@)
  {
    my($nntp, $group_name)=@_;
    process_group($nntp, $group_name, \&process_article);
  }

sub process_group(@)
{
  my($nntp, $group_name, $articlehandler)=@_;

  print "Processing group $group_name.\n";

  my ($total_articles, $firstarticle, $lastarticle) = $nntp->group($group_name);
  my $watermarkfile = $group_name . ".watermark" ;
  my $watermark     = read_watermark($watermarkfile);
  if($watermark > $firstarticle)
    {
      $firstarticle = $watermark+1;
    } 
  
  if($firstarticle <= $lastarticle)
    {
      print "Fetching " . ($lastarticle - $firstarticle) . " articles.\n";

      for(my $i=$firstarticle; $i<=$lastarticle; $i++)
        {
          print "\#$i: ";
          &$articlehandler($i, $nntp->article($i));
        }
      write_watermark($watermarkfile, $lastarticle);
    }
  else
    {
      print "no new articles\n";
    }
}

sub read_watermark($)
{
  open(WMF,"< $_[0]") or return -1;
  $wm = <WMF>;
  close WMF;
  chop $wm;
  return $wm;
}               

sub write_watermark(@)
{
  open(WMF,"> $_[0]") or die $!;
  print WMF "$_[1]\n";
  close WMF;
}

sub process_article($)
{
  my($n, $a)=@_;
  
  if($a == undef)
    {
      print "article expired\n";
    }
  else
    {
      print "$#$a lines";
      
      my $parser  = new MIME::Parser;
      $parser->output_to_core(1);      

      my $msg     = $parser->parse_data($a);
      #$msg->remove_sig(10);
      
      my $hdr     = $msg->head();
      $hdr->unfold;
      $hdr->decode;

      my $msgid       = trim($hdr->get("Message-ID"));
  
      $select_email_id_by_msgid->execute($msgid);
      my $ref = $select_email_id_by_msgid->fetchrow_hashref();
      $select_email_id_by_msgid->finish();
      if($ref)
        {
          print " - DUPLICATE\n";
        } 
      else
        {
          my @from        = Mail::Address->parse(trim($hdr->get('From')));
          my $sender_addr = $from[0]->address();
          
          if(blacklisted($sender_addr, \@articles_blacklist))
            {
              print " - BLACKLISTED ($sender_addr)\n";
            } 
          else 
            {
              if($body =~ /^\s+no forward/i)
                {
                  print " - NO FORWARD ($sender_addr)\n";
                } 
              else
                {
                  print " - NEW ($sender_addr)\n";
                  
                  my $sender_name = $from[0]->name();
                  my $sender_id   = add_author($sender_name, $sender_addr);
                  my $body        = fix_encoding($msg->bodyhandle()->as_string()); 
                  my $subject     = trim($hdr->get("Subject"));
                  my $ftnmsgid    = trim($hdr->get("X-FTN-MSGID"));
                  my $when        = parse_msg_date(trim($hdr->get("Date")));
                  
                  add_text($n, $sender_id, $when, $body, $msgid, $ftnmsgid, $subject, $hdr->as_string());
                }
              }
        }
      $msg->purge;
    }
}

sub fix_encoding
{
    $Lingua::DetectCharset::DoubtLog=0;
    my $charset = Lingua::DetectCharset::Detect($_[0]);
    if(($charset eq $DBCHARSET) || ($charset eq 'ENG'))
      {
        return $_[0];
      } 
    else
      {
        my $res = Convert::Cyrillic::cstocs($charset, $DBCHARSET, $_[0]);
        return $res;
      }
}

sub add_text
{
    my ($n, $sender_id, $when, $body, $msgid, $ftnmsgid, $subject, $hdr)=@_;

    $insert_email->execute($msgid, $ftnmsgid, $hdr);
    my $email_id = $insert_email->{"mysql_insertid"};
    $insert_email->finish();

    $insert_text->execute($sender_id, $when, $email_id, $subject, length($body), $body, 0);
    $insert_text->finish();
}

sub add_author
{
    my ($name, $email)=@_;
    my $id;
    
    $select_author_id_by_email->execute($email);
    if(my $ref = $select_author_id_by_email->fetchrow_hashref())
      {
        $id = $ref->{'Id'};
      } 
    else
      {
        $insert_author->execute($name, $email);
        $id = $insert_author->{"mysql_insertid"};
        $insert_author->finish();
      }
    $select_author_id_by_email->finish();
    return $id;
}

sub blacklisted(@)
{
  my($email, $list) = @_;
  
  foreach my $pat (@$list)
    {
      if($email =~ /$pat/)
        {
          return 1;
        }
    }
  
  return 0;
}

sub process_comment($)
{
  my($n, $a)=@_;
  
  if($a == undef)
    {
      print "comment expired\n";
    }
  else
    {
      print "$#$a lines";
      
      my $parser  = new MIME::Parser;
      $parser->output_to_core(1);      

      my $msg     = $parser->parse_data($a);
      #$msg->remove_sig(10);
      
      my $hdr     = $msg->head();
      $hdr->unfold;
      $hdr->decode;

      my $msgid       = trim($hdr->get("Message-ID"));
  
      $select_email_id_by_msgid->execute($msgid);
      my $ref = $select_email_id_by_msgid->fetchrow_hashref();
      $select_email_id_by_msgid->finish();
      if($ref)
        {
          print " - DUPLICATE\n";
        } 
      else
        {
          my @from        = Mail::Address->parse(trim($hdr->get('From')));
          my $sender_addr = $from[0]->address();
          
          if(blacklisted($sender_addr, \@comments_blacklist))
            {
              print " - BLACKLISTED ($sender_addr)\n";
            } 
          else
            {
              print " - NEW ($sender_addr)";
              
              my $sender_name = fix_encoding($from[0]->name());
              my $sender_id   = add_author($sender_name, $sender_addr);
              my $body        = fix_encoding($msg->bodyhandle()->as_string()); 
              my $subject     = fix_encoding(trim($hdr->get("Subject")));
              my $ftnmsgid    = trim($hdr->get("X-FTN-MSGID"));
              
              my $ftnreply    = trim($hdr->get("X-FTN-REPLY"));
              my $references  = trim($hdr->get("References"));
              my $when        = parse_msg_date(trim($hdr->get("Date")));
              
              add_comment($n, $sender_id, $when, $body, $msgid, $ftnmsgid, $subject, $hdr->as_string(), $ftnreply, $references);
              print "\n";
            }
        }
      $msg->purge;
    }
}

# Parse RFC822 'References' field 
# and returns all found message ids.
sub parse_references($)
{
    my $str = $_[0] ;
    my $i   = 0     ;
    my @res         ; 
    
    $_=$str;
    while(/(\<[^\s]+\@[^\s]+\>)/g)
      {
        @res[$i++]=$1;
      }
    return @res;
  }

sub add_comment
  {
    my($n, $sender_id, $when, $body, $msgid, $ftnmsgid, $subject, $hdr, $ftnreply, $references) = @_;
    
    # starting new thread, not associated to any next
    my $thread_id  = start_new_thread($subject);
    
    # Log unresolved refernces
    
    if($references)
      {
        my @refs = parse_references($references);
        foreach my $ref (@refs)
          {
            $insert_unsolved_msgid->execute($thread_id, $ref);
            $insert_unsolved_msgid->finish();
          }
      }
    
    if($ftnreply)
      {
        $insert_unsolved_ftnmsgid->execute($thread_id, $ftnreply);
        $insert_unsolved_ftnmsgid->finish();
      }
    
    $insert_email->execute($msgid, $ftnmsgid, $hdr);
    my $new_email_id = $insert_email->{"mysql_insertid"};
    $insert_email->finish();
    
    $insert_comment->execute($sender_id, $when, $new_email_id, $subject, length($body), $body, $thread_id);
    $insert_comment->finish();
}

sub start_new_thread
{
  my($title) = @_;
  
  $start_thread->execute($title);
  my $res = $start_thread->{"mysql_insertid"};
  print "Created thread with ID: $res\n";
  $start_thread->finish();
  return $res;
}

sub mark_thread_as_linked
{
  my($child_id) = @_;

  $remove_unsolved_ftnmsgid->execute($child_id);
  $remove_unsolved_ftnmsgid->finish();

  $remove_unsolved_msgid->execute($child_id);
  $remove_unsolved_msgid->finish();
}


sub iterate_statement
{
  my($stmt, $method) = @_;
  my $clink, $more;

  $stmt->execute();
  if($clink = $stmt->fetchrow_hashref())
    {
      &$method($clink);
      $more=1;
    } 
  else
    {
      $more=0;
    }
  $stmt->finish();
  return $more;
}

sub find_threads($)
{
  my($keep_days)=@_;

  print "Re-linking old threads\n";
  
  # -- Link comments to texts (new threads)

  while(iterate_statement($comments_to_texts_by_ftnmsgid ,\&link_text_to_comment)) {};
  while(iterate_statement($comments_to_texts_by_msgid    ,\&link_text_to_comment)) {};

  # -- Link comments to texts (merge threads)

  while(iterate_statement($comments_to_texts_add_by_ftnmsgid ,\&add_text_to_comment)) {};
  while(iterate_statement($comments_to_texts_add_by_msgid    ,\&add_text_to_comment)) {};

  # -- Link comments to comments

  while(iterate_statement($comments_to_comments_by_ftnmsgid ,\&link_2_comments)) {};
  while(iterate_statement($comments_to_comments_by_msgid    ,\&link_2_comments)) {};

  # -- Link comments to comments which both refer to non-existing parent
  
  while(iterate_statement($unresolved_comments_to_comment_by_ftnmsgid , \&link_2_unresolved_comments)) {};
  while(iterate_statement($unresolved_comments_to_comment_by_msgid    , \&link_2_unresolved_comments)) {};

  # -- delete duplicate unsolved references produced by previous step

  while(iterate_statement($find_ftn_dupes , \&remove_ftn_dupe)) {};
  while(iterate_statement($find_msg_dupes , \&remove_msg_dupe)) {};

  # drop empty threads
  $remove_empty_threads->execute();
  while(my $thr = $remove_empty_threads->fetchrow_hashref())
    {
      $delete_thread->execute($thr->{'Id'});
      $delete_thread->finish();
    }
  $remove_empty_threads->finish();

  # -- delete old unsolved message id references

  print "Removing unsolved references older than $keep_days days.\n";
  
  $remove_old_msg_unsolved->execute($keep_days);
  $remove_old_msg_unsolved->finish();

  $remove_old_ftn_unsolved->execute($keep_days);
  $remove_old_ftn_unsolved->finish();

}

sub remove_ftn_dupe
{
  my($clink) = @_;
  
  my $thread_id  = $clink->{'ThreadId'  };
  my $ftnmsg_id  = $clink->{'FTNMsgId'  };

  $delete_ftn_dupe->execute($thread_id, $ftnmsg_id);
  $delete_ftn_dupe->finish();

  print "Removing duplicate unsolved reference ($thread_id,\"$ftnmsg_id\")\n";
}

sub remove_msg_dupe
{
  my($clink) = @_;
  
  my $thread_id  = $clink->{'ThreadId'  };
  my $msg_id     = $clink->{'MsgId'     };

  $delete_msg_dupe->execute($thread_id, $msg_id);
  $delete_msg_dupe->finish();

  print "Removing duplicate unsolved reference ($thread_id,\"$msg_id\")\n";
}

sub link_2_comments
{
  my($clink) = @_;
  
  my $parent_id  = $clink->{'Id'             };
  my $thread_id  = $clink->{'ThreadId'       };
  my $parent_pos = $clink->{'ThreadPosition' };
  my $parent_off = $clink->{'ThreadOffset'   };
  my $child_id   = $clink->{'CThreadId'      };

  print "Linking thread #$child_id to thread #$thread_id at comment #$parent_id\n";
  
  merge_2_threads($thread_id, $child_id, $parent_pos, $parent_off);
  mark_thread_as_linked($child_id);
}

sub link_2_unresolved_comments
{
  my($clink) = @_;
  
  my $thread_id  = $clink->{'ThreadId'       };
  my $child_id   = $clink->{'CThreadId'      };
  
  merge_2_threads($thread_id, $child_id, -1, -1);

  $change_unresolved_ftn_thread->execute($thread_id, $child_id);
  $change_unresolved_ftn_thread->finish();
  $change_unresolved_msg_thread->execute($thread_id, $child_id);
  $change_unresolved_msg_thread->finish();

  print "Merging threads #$child_id and thread #$thread_id since they have same (unresolved) parent\n";

  mark_thread_as_linked($child_id);
}

sub link_text_to_comment
{
  my($clink) = @_;

  my $text_id    = $clink->{'Id'       };
  my $thread_id  = $clink->{'ThreadId' };

  $set_new_texts_thread_by_text_id->execute($thread_id, $text_id);
  $set_new_texts_thread_by_text_id->finish();

  print "Linking thread #$thread_id to text #$text_id as new comments thread\n";
  
  mark_thread_as_linked($thread_id);
}

sub add_text_to_comment
{
  my($clink) = @_;

  my $text_id   = $clink->{'Id' };
  my $thread_id = $clink->{'CommentThreadId' };
  my $child_id  = $clink->{'ThreadId'         };

  print "Linking thread #$child_id to thread #$thread_id for text #$text_id\n";

  merge_2_threads($thread_id, $child_id, -1, -1);
  mark_thread_as_linked($child_id);
}

sub merge_2_threads
{
  my($thread_id, $child_id, $parent_pos, $parent_off) = @_;

  # find first comment before which new comments must be inserted
  $find_comment_merge_point->execute($parent_pos, $parent_off, $thread_id, $child_id);
  my $res = $find_comment_merge_point->fetchrow_hashref();
  $find_comment_merge_point->finish();

  my $merge_off=$parent_off+1;
  my $merge_pos;

  if($res)
    {
      # inserting before existing comment in same thread on same level
      $merge_pos = $res->{'ThreadPosition'};
    }
  else
    {
      # more recent than any comment, insert at the end of comments thread
      $find_comment_merge_point_next_parent->execute($parent_pos, $parent_off, $thread_id);
      ($merge_pos) = $find_comment_merge_point_next_parent->fetchrow_array();
      $find_comment_merge_point_next_parent->finish();
      if($merge_pos)
        {
          # inserting instead next comment on parent (or higher) level.
          $merge_pos--;
        }
      else
        {
          # inserting at the very end of the list
          $find_comment_merge_point_last_response->execute($parent_pos, $thread_id);
          ($merge_pos) = $find_comment_merge_point_last_response->fetchrow_array();
          $find_comment_merge_point_last_response->finish();
          if(!$merge_pos)
            {
              print "[ERROR: no comments to merge!]";
            }
        }
    }
  
  # do actual merge

  $count_comments_in_thread->execute($child_id);
  my($delta) = $count_comments_in_thread->fetchrow_array();
  $count_comments_in_thread->finish();
  
  # Make space right after parent
  $shuffle_thread->execute($delta, $thread_id, $merge_pos);
  $shuffle_thread->finish();
  
  # merge right after given position
  $link_thread_to_comment->execute($thread_id, $merge_off, $merge_pos, $child_id);
  $link_thread_to_comment->finish();

  # update thread creation date
  $oldest_in_thread->execute($thread_id);
  my($oldest)=$oldest_in_thread->fetchrow_array();
  $oldest_in_thread->finish();
  if($oldest)
    {
      $set_thread_date->execute($oldest, $thread_id);
    } 
  else
    {
      print "ERROR: merging empty threads! Can't get oldest messsage date for $thread_id\n";
    }
}

sub parse_msg_date
  {
  my $str = $_[0] ;
  $time = str2time($str);
  if(!$time)
    {
      # parsing failed. Assume NOW.
      print "Parsing of '$str' failed. Using current time instead.\n";
      $time = time();
    }
  
  return strftime "%Y-%m-%d %H:%M:%S", gmtime($time);
}
