sup

A curses threads-with-tags style email client

sup.git

git clone https://supmua.dev/git/sup/

lib/sup/index.rb (28918B) - raw

      1 ENV["XAPIAN_FLUSH_THRESHOLD"] = "1000"
      2 ENV["XAPIAN_CJK_NGRAM"] = "1"
      3 
      4 require 'xapian'
      5 require 'set'
      6 require 'fileutils'
      7 require 'monitor'
      8 require 'chronic'
      9 
     10 require "sup/util/query"
     11 require "sup/interactive_lock"
     12 require "sup/hook"
     13 require "sup/logger/singleton"
     14 
     15 
     16 if ([Xapian.major_version, Xapian.minor_version, Xapian.revision] <=> [1,2,15]) < 0
     17   fail <<-EOF
     18 \n
     19 Xapian version 1.2.15 or higher required.
     20 If you have xapian-full-alaveteli installed,
     21 Please remove it by running `gem uninstall xapian-full-alaveteli`
     22 since it's been replaced by the xapian-ruby gem.
     23 
     24   EOF
     25 end
     26 
     27 module Redwood
     28 
     29 # This index implementation uses Xapian for searching and storage. It
     30 # tends to be slightly faster than Ferret for indexing and significantly faster
     31 # for searching due to precomputing thread membership.
     32 class Index
     33   include InteractiveLock
     34 
     35   INDEX_VERSION = '4'
     36 
     37   ## dates are converted to integers for xapian, and are used for document ids,
     38   ## so we must ensure they're reasonably valid. this typically only affect
     39   ## spam.
     40   MIN_DATE = Time.at 0
     41   MAX_DATE = Time.at(2**31-1)
     42 
     43   HookManager.register "custom-search", <<EOS
     44 Executes before a string search is applied to the index,
     45 returning a new search string.
     46 Variables:
     47   subs: The string being searched.
     48 EOS
     49 
     50   class LockError < StandardError
     51     def initialize h
     52       @h = h
     53     end
     54 
     55     def method_missing m; @h[m.to_s] end
     56   end
     57 
     58   include Redwood::Singleton
     59 
     60   def initialize dir=BASE_DIR
     61     @dir = dir
     62     FileUtils.mkdir_p @dir
     63     @lock = Lockfile.new lockfile, :retries => 0, :max_age => nil
     64     @sync_worker = nil
     65     @sync_queue = Queue.new
     66     @index_mutex = Monitor.new
     67   end
     68 
     69   def lockfile; File.join @dir, "lock" end
     70 
     71   def lock
     72     debug "locking #{lockfile}..."
     73     begin
     74       @lock.lock
     75     rescue Lockfile::MaxTriesLockError
     76       raise LockError, @lock.lockinfo_on_disk
     77     end
     78   end
     79 
     80   def start_lock_update_thread
     81     @lock_update_thread = Redwood::reporting_thread("lock update") do
     82       while true
     83         sleep 30
     84         @lock.touch_yourself
     85       end
     86     end
     87   end
     88 
     89   def stop_lock_update_thread
     90     @lock_update_thread.kill if @lock_update_thread
     91     @lock_update_thread = nil
     92   end
     93 
     94   def unlock
     95     if @lock && @lock.locked?
     96       debug "unlocking #{lockfile}..."
     97       @lock.unlock
     98     end
     99   end
    100 
    101   def load failsafe=false
    102     SourceManager.load_sources File.join(@dir, "sources.yaml")
    103     load_index failsafe
    104   end
    105 
    106   def save
    107     debug "saving index and sources..."
    108     FileUtils.mkdir_p @dir unless File.exist? @dir
    109     SourceManager.save_sources File.join(@dir, "sources.yaml")
    110     save_index
    111   end
    112 
    113   def get_xapian
    114     @xapian
    115   end
    116 
    117   def load_index failsafe=false
    118     path = File.join(@dir, 'xapian')
    119     if File.exist? path
    120       @xapian = Xapian::WritableDatabase.new(path, Xapian::DB_OPEN)
    121       db_version = @xapian.get_metadata 'version'
    122       db_version = '0' if db_version.empty?
    123       if false
    124         info "Upgrading index format #{db_version} to #{INDEX_VERSION}"
    125         @xapian.set_metadata 'version', INDEX_VERSION
    126       elsif db_version != INDEX_VERSION
    127         fail "This Sup version expects a v#{INDEX_VERSION} index, but you have an existing v#{db_version} index. Please run sup-dump to save your labels, move #{path} out of the way, and run sup-sync --restore."
    128       end
    129     else
    130       @xapian = Xapian::WritableDatabase.new(path, Xapian::DB_CREATE)
    131       @xapian.set_metadata 'version', INDEX_VERSION
    132       @xapian.set_metadata 'rescue-version', '0'
    133     end
    134     @enquire = Xapian::Enquire.new @xapian
    135     @enquire.weighting_scheme = Xapian::BoolWeight.new
    136     @enquire.docid_order = Xapian::Enquire::ASCENDING
    137   end
    138 
    139   def add_message m; sync_message m, true end
    140   def update_message m; sync_message m, true end
    141   def update_message_state m; sync_message m[0], false, m[1] end
    142 
    143   def save_index
    144     info "Flushing Xapian updates to disk. This may take a while..."
    145     @xapian.commit
    146   end
    147 
    148   def contains_id? id
    149     synchronize { find_docid(id) && true }
    150   end
    151 
    152   def contains? m; contains_id? m.id end
    153 
    154   def size
    155     synchronize { @xapian.doccount }
    156   end
    157 
    158   def empty?; size == 0 end
    159 
    160   ## Yields a message-id and message-building lambda for each
    161   ## message that matches the given query, in descending date order.
    162   ## You should probably not call this on a block that doesn't break
    163   ## rather quickly because the results can be very large.
    164   def each_id_by_date query={}
    165     each_id(query) { |id| yield id, lambda { build_message id } }
    166   end
    167 
    168   ## Return the number of matches for query in the index
    169   def num_results_for query={}
    170     xapian_query = build_xapian_query query
    171     matchset = run_query xapian_query, 0, 0, 100
    172     matchset.matches_estimated
    173   end
    174 
    175   ## check if a message is part of a killed thread
    176   ## (warning: duplicates code below)
    177   ## NOTE: We can be more efficient if we assume every
    178   ## killed message that hasn't been initially added
    179   ## to the indexi s this way
    180   def message_joining_killed? m
    181     return false unless doc = find_doc(m.id)
    182     queue = doc.value(THREAD_VALUENO).split(',')
    183     seen_threads = Set.new
    184     seen_messages = Set.new [m.id]
    185     while not queue.empty?
    186       thread_id = queue.pop
    187       next if seen_threads.member? thread_id
    188       return true if thread_killed?(thread_id)
    189       seen_threads << thread_id
    190       docs = term_docids(mkterm(:thread, thread_id)).map { |x| @xapian.document x }
    191       docs.each do |doc|
    192         msgid = doc.value MSGID_VALUENO
    193         next if seen_messages.member? msgid
    194         seen_messages << msgid
    195         queue.concat doc.value(THREAD_VALUENO).split(',')
    196       end
    197     end
    198     false
    199   end
    200 
    201   ## yield all messages in the thread containing 'm' by repeatedly
    202   ## querying the index. yields pairs of message ids and
    203   ## message-building lambdas, so that building an unwanted message
    204   ## can be skipped in the block if desired.
    205   ##
    206   ## only two options, :limit and :skip_killed. if :skip_killed is
    207   ## true, stops loading any thread if a message with a :killed flag
    208   ## is found.
    209   def each_message_in_thread_for m, opts={}
    210     # TODO thread by subject
    211     return unless doc = find_doc(m.id)
    212     queue = doc.value(THREAD_VALUENO).split(',')
    213     msgids = [m.id]
    214     seen_threads = Set.new
    215     seen_messages = Set.new [m.id]
    216     while not queue.empty?
    217       thread_id = queue.pop
    218       next if seen_threads.member? thread_id
    219       return false if opts[:skip_killed] && thread_killed?(thread_id)
    220       seen_threads << thread_id
    221       docs = term_docids(mkterm(:thread, thread_id)).map { |x| @xapian.document x }
    222       docs.each do |doc|
    223         msgid = doc.value MSGID_VALUENO
    224         next if seen_messages.member? msgid
    225         msgids << msgid
    226         seen_messages << msgid
    227         queue.concat doc.value(THREAD_VALUENO).split(',')
    228       end
    229     end
    230     msgids.each { |id| yield id, lambda { build_message id } }
    231     true
    232   end
    233 
    234   ## Load message with the given message-id from the index
    235   def build_message id
    236     entry = synchronize { get_entry id }
    237     return unless entry
    238 
    239     locations = entry[:locations].map do |source_id,source_info|
    240       source = SourceManager[source_id]
    241       raise "invalid source #{source_id}" unless source
    242       Location.new source, source_info
    243     end
    244 
    245     m = Message.new :locations => locations,
    246                     :labels => entry[:labels],
    247                     :snippet => entry[:snippet]
    248 
    249     # Try to find person from contacts before falling back to
    250     # generating it from the address.
    251     mk_person = lambda { |x| Person.from_name_and_email(*x.reverse!) }
    252     entry[:from] = mk_person[entry[:from]]
    253     entry[:to].map!(&mk_person)
    254     entry[:cc].map!(&mk_person)
    255     entry[:bcc].map!(&mk_person)
    256 
    257     m.load_from_index! entry
    258     m
    259   end
    260 
    261   ## Delete message with the given message-id from the index
    262   def delete id
    263     synchronize { @xapian.delete_document mkterm(:msgid, id) }
    264   end
    265 
    266   ## Given an array of email addresses, return an array of Person objects that
    267   ## have sent mail to or received mail from any of the given addresses.
    268   def load_contacts email_addresses, opts={}
    269     contacts = Set.new
    270     num = opts[:num] || 20
    271     each_id_by_date :participants => email_addresses do |id,b|
    272       break if contacts.size >= num
    273       m = b.call
    274       ([m.from]+m.to+m.cc+m.bcc).compact.each { |p| contacts << [p.name, p.email] }
    275     end
    276     contacts.to_a.compact[0...num].map { |n,e| Person.from_name_and_email n, e }
    277   end
    278 
    279   ## Yield each message-id matching query
    280   EACH_ID_PAGE = 100
    281   def each_id query={}, ignore_neg_terms = true
    282     offset = 0
    283     page = EACH_ID_PAGE
    284 
    285     xapian_query = build_xapian_query query, ignore_neg_terms
    286     while true
    287       ids = run_query_ids xapian_query, offset, (offset+page)
    288       ids.each { |id| yield id }
    289       break if ids.size < page
    290       offset += page
    291     end
    292   end
    293 
    294   ## Yield each message matching query
    295   ## The ignore_neg_terms parameter is used to display result even if
    296   ## it contains "forbidden" labels such as :deleted, it is used in
    297   ## Poll#poll_from when we need to get the location of a message that
    298   ## may contain these labels
    299   def each_message query={}, ignore_neg_terms = true, &b
    300     each_id query, ignore_neg_terms do |id|
    301       yield build_message(id)
    302     end
    303   end
    304 
    305   # Search messages. Returns an Enumerator.
    306   def find_messages query_expr
    307     enum_for :each_message, parse_query(query_expr)
    308   end
    309 
    310   # wrap all future changes inside a transaction so they're done atomically
    311   def begin_transaction
    312     synchronize { @xapian.begin_transaction }
    313   end
    314 
    315   # complete the transaction and write all previous changes to disk
    316   def commit_transaction
    317     synchronize { @xapian.commit_transaction }
    318   end
    319 
    320   # abort the transaction and revert all changes made since begin_transaction
    321   def cancel_transaction
    322     synchronize { @xapian.cancel_transaction }
    323   end
    324 
    325   ## xapian-compact takes too long, so this is a no-op
    326   ## until we think of something better
    327   def optimize
    328   end
    329 
    330   ## Return the id source of the source the message with the given message-id
    331   ## was synced from
    332   def source_for_id id
    333     synchronize { get_entry(id)[:source_id] }
    334   end
    335 
    336   ## Yields each term in the index that starts with prefix
    337   def each_prefixed_term prefix
    338     term = @xapian._dangerous_allterms_begin prefix
    339     lastTerm = @xapian._dangerous_allterms_end prefix
    340     until term.equals lastTerm
    341       yield term.term
    342       term.next
    343     end
    344     nil
    345   end
    346 
    347   ## Yields (in lexicographical order) the source infos of all locations from
    348   ## the given source with the given source_info prefix
    349   def each_source_info source_id, prefix='', &b
    350     p = mkterm :location, source_id, prefix
    351     each_prefixed_term p do |x|
    352       yield prefix + x[p.length..-1]
    353     end
    354   end
    355 
    356   class ParseError < StandardError; end
    357 
    358   # Stemmed
    359   NORMAL_PREFIX = {
    360     'subject' => {:prefix => 'S', :exclusive => false},
    361     'body' => {:prefix => 'B', :exclusive => false},
    362     'from_name' => {:prefix => 'FN', :exclusive => false},
    363     'to_name' => {:prefix => 'TN', :exclusive => false},
    364     'name' => {:prefix => %w(FN TN), :exclusive => false},
    365     'attachment' => {:prefix => 'A', :exclusive => false},
    366     'email_text' => {:prefix => 'E', :exclusive => false},
    367     '' => {:prefix => %w(S B FN TN A E), :exclusive => false},
    368   }
    369 
    370   # Unstemmed
    371   BOOLEAN_PREFIX = {
    372     'type' => {:prefix => 'K', :exclusive => true},
    373     'from_email' => {:prefix => 'FE', :exclusive => false},
    374     'to_email' => {:prefix => 'TE', :exclusive => false},
    375     'email' => {:prefix => %w(FE TE), :exclusive => false},
    376     'date' => {:prefix => 'D', :exclusive => true},
    377     'label' => {:prefix => 'L', :exclusive => false},
    378     'source_id' => {:prefix => 'I', :exclusive => true},
    379     'attachment_extension' => {:prefix => 'O', :exclusive => false},
    380     'msgid' => {:prefix => 'Q', :exclusive => true},
    381     'id' => {:prefix => 'Q', :exclusive => true},
    382     'thread' => {:prefix => 'H', :exclusive => false},
    383     'ref' => {:prefix => 'R', :exclusive => false},
    384     'location' => {:prefix => 'J', :exclusive => false},
    385   }
    386 
    387   PREFIX = NORMAL_PREFIX.merge BOOLEAN_PREFIX
    388 
    389   COMPL_OPERATORS = %w[AND OR NOT]
    390   COMPL_PREFIXES = (
    391     %w[
    392       from to
    393       is has label
    394       filename filetypem
    395       before on in during after
    396       limit
    397     ] + NORMAL_PREFIX.keys + BOOLEAN_PREFIX.keys
    398   ).map{|p|"#{p}:"} + COMPL_OPERATORS
    399 
    400   ## parse a query string from the user. returns a query object
    401   ## that can be passed to any index method with a 'query'
    402   ## argument.
    403   ##
    404   ## raises a ParseError if something went wrong.
    405   def parse_query s
    406     query = {}
    407 
    408     subs = HookManager.run("custom-search", :subs => s) || s
    409     begin
    410       subs = SearchManager.expand subs
    411     rescue SearchManager::ExpansionError => e
    412       raise ParseError, e.message
    413     end
    414     subs = subs.gsub(/\b(to|from):(\S+)\b/) do
    415       field, value = $1, $2
    416       email_field, name_field = %w(email name).map { |x| "#{field}_#{x}" }
    417       if(p = ContactManager.contact_for(value))
    418         "#{email_field}:#{p.email}"
    419       elsif value == "me"
    420         '(' + AccountManager.user_emails.map { |e| "#{email_field}:#{e}" }.join(' OR ') + ')'
    421       else
    422         "(#{email_field}:#{value} OR #{name_field}:#{value})"
    423       end
    424     end
    425 
    426     ## gmail style "is" operator
    427     subs = subs.gsub(/\b(is|has):(\S+)\b/) do
    428       _field, label = $1, $2
    429       case label
    430       when "read"
    431         "-label:unread"
    432       when "spam"
    433         query[:load_spam] = true
    434         "label:spam"
    435       when "deleted"
    436         query[:load_deleted] = true
    437         "label:deleted"
    438       else
    439         "label:#{$2}"
    440       end
    441     end
    442 
    443     ## labels are stored lower-case in the index
    444     subs = subs.gsub(/\blabel:(\S+)\b/) do
    445       label = $1
    446       "label:#{label.downcase}"
    447     end
    448 
    449     ## if we see a label:deleted or a label:spam term anywhere in the query
    450     ## string, we set the extra load_spam or load_deleted options to true.
    451     ## bizarre? well, because the query allows arbitrary parenthesized boolean
    452     ## expressions, without fully parsing the query, we can't tell whether
    453     ## the user is explicitly directing us to search spam messages or not.
    454     ## e.g. if the string is -(-(-(-(-label:spam)))), does the user want to
    455     ## search spam messages or not?
    456     ##
    457     ## so, we rely on the fact that turning these extra options ON turns OFF
    458     ## the adding of "-label:deleted" or "-label:spam" terms at the very
    459     ## final stage of query processing. if the user wants to search spam
    460     ## messages, not adding that is the right thing; if he doesn't want to
    461     ## search spam messages, then not adding it won't have any effect.
    462     query[:load_spam] = true if subs =~ /\blabel:spam\b/
    463     query[:load_deleted] = true if subs =~ /\blabel:deleted\b/
    464     query[:load_killed] = true if subs =~ /\blabel:killed\b/
    465 
    466     ## gmail style attachments "filename" and "filetype" searches
    467     subs = subs.gsub(/\b(filename|filetype):(\((.+?)\)\B|(\S+)\b)/) do
    468       field, name = $1, ($3 || $4)
    469       case field
    470       when "filename"
    471         debug "filename: translated #{field}:#{name} to attachment:\"#{name.downcase}\""
    472         "attachment:\"#{name.downcase}\""
    473       when "filetype"
    474         debug "filetype: translated #{field}:#{name} to attachment_extension:#{name.downcase}"
    475         "attachment_extension:#{name.downcase}"
    476       end
    477     end
    478 
    479     lastdate = 2<<32 - 1
    480     firstdate = 0
    481     subs = subs.gsub(/\b(before|on|in|during|after):(\((.+?)\)\B|(\S+)\b)/) do
    482       field, datestr = $1, ($3 || $4)
    483       realdate = Chronic.parse datestr, :guess => false, :context => :past
    484       if realdate
    485         case field
    486         when "after"
    487           debug "chronic: translated #{field}:#{datestr} to #{realdate.end}"
    488           "date:#{realdate.end.to_i}..#{lastdate}"
    489         when "before"
    490           debug "chronic: translated #{field}:#{datestr} to #{realdate.begin}"
    491           "date:#{firstdate}..#{realdate.end.to_i}"
    492         else
    493           debug "chronic: translated #{field}:#{datestr} to #{realdate}"
    494           "date:#{realdate.begin.to_i}..#{realdate.end.to_i}"
    495         end
    496       else
    497         raise ParseError, "can't understand date #{datestr.inspect}"
    498       end
    499     end
    500 
    501     ## limit:42 restrict the search to 42 results
    502     subs = subs.gsub(/\blimit:(\S+)\b/) do
    503       lim = $1
    504       if lim =~ /^\d+$/
    505         query[:limit] = lim.to_i
    506         ''
    507       else
    508         raise ParseError, "non-numeric limit #{lim.inspect}"
    509       end
    510     end
    511 
    512     debug "translated query: #{subs.inspect}"
    513 
    514     qp = Xapian::QueryParser.new
    515     qp.database = @xapian
    516     qp.stemmer = Xapian::Stem.new($config[:stem_language])
    517     qp.stemming_strategy = Xapian::QueryParser::STEM_SOME
    518     qp.default_op = Xapian::Query::OP_AND
    519     begin
    520       rangeprocessor = Xapian::NumberRangeProcessor.new DATE_VALUENO, 'date:'
    521       qp.add_rangeprocessor rangeprocessor
    522     rescue NameError  # xapian < 1.3
    523       valuerangeprocessor = Xapian::NumberValueRangeProcessor.new DATE_VALUENO, 'date:', true
    524       qp.add_valuerangeprocessor valuerangeprocessor
    525     end
    526     NORMAL_PREFIX.each { |k,info| info[:prefix].each {
    527       |v| qp.add_prefix k, v }
    528     }
    529     BOOLEAN_PREFIX.each { |k,info| info[:prefix].each {
    530       |v| qp.add_boolean_prefix k, v, info[:exclusive] }
    531     }
    532 
    533     begin
    534       xapian_query = qp.parse_query(subs, Xapian::QueryParser::FLAG_PHRASE   |
    535                                           Xapian::QueryParser::FLAG_BOOLEAN  |
    536                                           Xapian::QueryParser::FLAG_LOVEHATE |
    537                                           Xapian::QueryParser::FLAG_WILDCARD)
    538     rescue RuntimeError => e
    539       raise ParseError, "xapian query parser error: #{e}"
    540     end
    541 
    542     debug "parsed xapian query: #{Util::Query.describe(xapian_query, subs)}"
    543 
    544     if xapian_query.nil? or xapian_query.empty?
    545       raise ParseError, "couldn't parse \"#{s}\" as xapian query " \
    546                         "(special characters aren't indexed)"
    547     end
    548 
    549     query[:qobj] = xapian_query
    550     query[:text] = s
    551     query
    552   end
    553 
    554   def save_message m, sync_back = true
    555     if @sync_worker
    556       @sync_queue << [m, sync_back]
    557     else
    558       update_message_state [m, sync_back]
    559     end
    560     m.clear_dirty
    561   end
    562 
    563   def save_thread t, sync_back = true
    564     t.each_dirty_message do |m|
    565       save_message m, sync_back
    566     end
    567   end
    568 
    569   def start_sync_worker
    570     @sync_worker = Redwood::reporting_thread('index sync') { run_sync_worker }
    571   end
    572 
    573   def stop_sync_worker
    574     return unless worker = @sync_worker
    575     @sync_worker = nil
    576     @sync_queue << :die
    577     worker.join
    578   end
    579 
    580   def run_sync_worker
    581     while m = @sync_queue.deq
    582       return if m == :die
    583       update_message_state m
    584       # Necessary to keep Xapian calls from lagging the UI too much.
    585       sleep 0.03
    586     end
    587   end
    588 
    589   private
    590 
    591   MSGID_VALUENO = 0
    592   THREAD_VALUENO = 1
    593   DATE_VALUENO = 2
    594 
    595   MAX_TERM_LENGTH = 245
    596 
    597   # Xapian can very efficiently sort in ascending docid order. Sup always wants
    598   # to sort by descending date, so this method maps between them. In order to
    599   # handle multiple messages per second, we use a logistic curve centered
    600   # around MIDDLE_DATE so that the slope (docid/s) is greatest in this time
    601   # period. A docid collision is not an error - the code will pick the next
    602   # smallest unused one.
    603   DOCID_SCALE = 2.0**32
    604   TIME_SCALE = 2.0**27
    605   MIDDLE_DATE = Time.gm(2011)
    606   def assign_docid m, truncated_date
    607     t = (truncated_date.to_i - MIDDLE_DATE.to_i).to_f
    608     docid = (DOCID_SCALE - DOCID_SCALE/(Math::E**(-(t/TIME_SCALE)) + 1)).to_i
    609     while docid > 0 and docid_exists? docid
    610       docid -= 1
    611     end
    612     docid > 0 ? docid : nil
    613   end
    614 
    615   # XXX is there a better way?
    616   def docid_exists? docid
    617     begin
    618       @xapian.doclength docid
    619       true
    620     rescue RuntimeError #Xapian::DocNotFoundError
    621       raise unless $!.message =~ /DocNotFoundError/
    622       false
    623     end
    624   end
    625 
    626   def term_docids term
    627     @xapian.postlist(term).map { |x| x.docid }
    628   end
    629 
    630   def find_docid id
    631     docids = term_docids(mkterm(:msgid,id))
    632     fail unless docids.size <= 1
    633     docids.first
    634   end
    635 
    636   def find_doc id
    637     return unless docid = find_docid(id)
    638     @xapian.document docid
    639   end
    640 
    641   def get_id docid
    642     return unless doc = @xapian.document(docid)
    643     doc.value MSGID_VALUENO
    644   end
    645 
    646   def get_entry id
    647     return unless doc = find_doc(id)
    648     doc.entry
    649   end
    650 
    651   def thread_killed? thread_id
    652     not run_query(Q.new(Q::OP_AND, mkterm(:thread, thread_id), mkterm(:label, :Killed)), 0, 1).empty?
    653   end
    654 
    655   def synchronize &b
    656     @index_mutex.synchronize(&b)
    657   end
    658 
    659   def run_query xapian_query, offset, limit, checkatleast=0
    660     synchronize do
    661       @enquire.query = xapian_query
    662       @enquire.mset(offset, limit-offset, checkatleast)
    663     end
    664   end
    665 
    666   def run_query_ids xapian_query, offset, limit
    667     matchset = run_query xapian_query, offset, limit
    668     matchset.matches.map { |r| r.document.value MSGID_VALUENO }
    669   end
    670 
    671   Q = Xapian::Query
    672   def build_xapian_query opts, ignore_neg_terms = true
    673     labels = ([opts[:label]] + (opts[:labels] || [])).compact
    674     neglabels = [:spam, :deleted, :killed].reject { |l| (labels.include? l) || opts.member?("load_#{l}".intern) }
    675     pos_terms, neg_terms = [], []
    676 
    677     pos_terms << mkterm(:type, 'mail')
    678     pos_terms.concat(labels.map { |l| mkterm(:label,l) })
    679     pos_terms << opts[:qobj] if opts[:qobj]
    680     pos_terms << mkterm(:source_id, opts[:source_id]) if opts[:source_id]
    681     pos_terms << mkterm(:location, *opts[:location]) if opts[:location]
    682 
    683     if opts[:participants]
    684       participant_terms = opts[:participants].map { |p| [:from,:to].map { |d| mkterm(:email, d, (Redwood::Person === p) ? p.email : p) } }.flatten
    685       pos_terms << Q.new(Q::OP_OR, participant_terms)
    686     end
    687 
    688     neg_terms.concat(neglabels.map { |l| mkterm(:label,l) }) if ignore_neg_terms
    689 
    690     pos_query = Q.new(Q::OP_AND, pos_terms)
    691     neg_query = Q.new(Q::OP_OR, neg_terms)
    692 
    693     if neg_query.empty?
    694       pos_query
    695     else
    696       Q.new(Q::OP_AND_NOT, [pos_query, neg_query])
    697     end
    698   end
    699 
    700   def sync_message m, overwrite, sync_back = true
    701     ## TODO: we should not save the message if the sync_back failed
    702     ## since it would overwrite the location field
    703     m.sync_back if sync_back
    704 
    705     doc = synchronize { find_doc(m.id) }
    706     existed = doc != nil
    707     doc ||= Xapian::Document.new
    708     do_index_static = overwrite || !existed
    709     old_entry = !do_index_static && doc.entry
    710     snippet = do_index_static ? m.snippet : old_entry[:snippet]
    711 
    712     entry = {
    713       :message_id => m.id,
    714       :locations => m.locations.map { |x| [x.source.id, x.info] },
    715       :date => truncate_date(m.date),
    716       :snippet => snippet,
    717       :labels => m.labels.to_a,
    718       :from => [m.from.email, m.from.name],
    719       :to => m.to.map { |p| [p.email, p.name] },
    720       :cc => m.cc.map { |p| [p.email, p.name] },
    721       :bcc => m.bcc.map { |p| [p.email, p.name] },
    722       :subject => m.subj,
    723       :refs => m.refs.to_a,
    724       :replytos => m.replytos.to_a,
    725     }
    726 
    727     if do_index_static
    728       doc.clear_terms
    729       doc.clear_values
    730       index_message_static m, doc, entry
    731     end
    732 
    733     index_message_locations doc, entry, old_entry
    734     index_message_threading doc, entry, old_entry
    735     index_message_labels doc, entry[:labels], (do_index_static ? [] : old_entry[:labels])
    736     doc.entry = entry
    737 
    738     synchronize do
    739       unless docid = existed ? doc.docid : assign_docid(m, truncate_date(m.date))
    740         # Could be triggered by spam
    741         warn "docid underflow, dropping #{m.id.inspect}"
    742         return
    743       end
    744       @xapian.replace_document docid, doc
    745     end
    746 
    747     m.labels.each { |l| LabelManager << l }
    748     true
    749   end
    750 
    751   ## Index content that can't be changed by the user
    752   def index_message_static m, doc, entry
    753     # Person names are indexed with several prefixes
    754     person_termer = lambda do |d|
    755       lambda do |p|
    756         doc.index_text p.name, PREFIX["#{d}_name"][:prefix] if p.name
    757         doc.index_text p.email, PREFIX['email_text'][:prefix]
    758         doc.add_term mkterm(:email, d, p.email)
    759       end
    760     end
    761 
    762     person_termer[:from][m.from] if m.from
    763     (m.to+m.cc+m.bcc).each(&(person_termer[:to]))
    764 
    765     # Full text search content
    766     subject_text = m.indexable_subject
    767     body_text = m.indexable_body
    768     doc.index_text subject_text, PREFIX['subject'][:prefix]
    769     doc.index_text body_text, PREFIX['body'][:prefix]
    770     m.attachments.each { |a| doc.index_text a, PREFIX['attachment'][:prefix] }
    771 
    772     # Miscellaneous terms
    773     doc.add_term mkterm(:date, m.date) if m.date
    774     doc.add_term mkterm(:type, 'mail')
    775     doc.add_term mkterm(:msgid, m.id)
    776     m.attachments.each do |a|
    777       a =~ /\.(\w+)$/ or next
    778       doc.add_term mkterm(:attachment_extension, $1)
    779     end
    780 
    781     # Date value for range queries
    782     date_value = begin
    783       Xapian.sortable_serialise m.date.to_i
    784     rescue TypeError
    785       Xapian.sortable_serialise 0
    786     end
    787 
    788     doc.add_value MSGID_VALUENO, m.id
    789     doc.add_value DATE_VALUENO, date_value
    790   end
    791 
    792   def index_message_locations doc, entry, old_entry
    793     old_entry[:locations].map { |x| x[0] }.uniq.each { |x| doc.remove_term mkterm(:source_id, x) } if old_entry
    794     entry[:locations].map { |x| x[0] }.uniq.each { |x| doc.add_term mkterm(:source_id, x) }
    795     old_entry[:locations].each { |x| (doc.remove_term mkterm(:location, *x) rescue nil) } if old_entry
    796     entry[:locations].each { |x| doc.add_term mkterm(:location, *x) }
    797   end
    798 
    799   def index_message_labels doc, new_labels, old_labels
    800     return if new_labels == old_labels
    801     added = new_labels.to_a - old_labels.to_a
    802     removed = old_labels.to_a - new_labels.to_a
    803     added.each { |t| doc.add_term mkterm(:label,t) }
    804     removed.each { |t| doc.remove_term mkterm(:label,t) }
    805   end
    806 
    807   ## Assign a set of thread ids to the document. This is a hybrid of the runtime
    808   ## search done by the Ferret index and the index-time union done by previous
    809   ## versions of the Xapian index. We first find the thread ids of all messages
    810   ## with a reference to or from us. If that set is empty, we use our own
    811   ## message id. Otherwise, we use all the thread ids we previously found. In
    812   ## the common case there's only one member in that set, but if we're the
    813   ## missing link between multiple previously unrelated threads we can have
    814   ## more. XapianIndex#each_message_in_thread_for follows the thread ids when
    815   ## searching so the user sees a single unified thread.
    816   def index_message_threading doc, entry, old_entry
    817     return if old_entry && (entry[:refs] == old_entry[:refs]) && (entry[:replytos] == old_entry[:replytos])
    818     children = term_docids(mkterm(:ref, entry[:message_id])).map { |docid| @xapian.document docid }
    819     parent_ids = entry[:refs] + entry[:replytos]
    820     parents = parent_ids.map { |id| find_doc id }.compact
    821     thread_members = SavingHash.new { [] }
    822     (children + parents).each do |doc2|
    823       thread_ids = doc2.value(THREAD_VALUENO).split ','
    824       thread_ids.each { |thread_id| thread_members[thread_id] << doc2 }
    825     end
    826     thread_ids = thread_members.empty? ? [entry[:message_id]] : thread_members.keys
    827     thread_ids.each { |thread_id| doc.add_term mkterm(:thread, thread_id) }
    828     parent_ids.each { |ref| doc.add_term mkterm(:ref, ref) }
    829     doc.add_value THREAD_VALUENO, (thread_ids * ',')
    830   end
    831 
    832   def truncate_date date
    833     if date < MIN_DATE
    834       debug "warning: adjusting too-low date #{date} for indexing"
    835       MIN_DATE
    836     elsif date > MAX_DATE
    837       debug "warning: adjusting too-high date #{date} for indexing"
    838       MAX_DATE
    839     else
    840       date
    841     end
    842   end
    843 
    844   # Construct a Xapian term
    845   def mkterm type, *args
    846     case type
    847     when :label
    848       PREFIX['label'][:prefix] + args[0].to_s.downcase
    849     when :type
    850       PREFIX['type'][:prefix] + args[0].to_s.downcase
    851     when :date
    852       PREFIX['date'][:prefix] + args[0].getutc.strftime("%Y%m%d%H%M%S")
    853     when :email
    854       case args[0]
    855       when :from then PREFIX['from_email'][:prefix]
    856       when :to then PREFIX['to_email'][:prefix]
    857       else raise "Invalid email term type #{args[0]}"
    858       end + args[1].to_s.downcase
    859     when :source_id
    860       PREFIX['source_id'][:prefix] + args[0].to_s.downcase
    861     when :location
    862       PREFIX['location'][:prefix] + [args[0]].pack('n') + args[1].to_s
    863     when :attachment_extension
    864       PREFIX['attachment_extension'][:prefix] + args[0].to_s.downcase
    865     when :msgid, :ref, :thread
    866       PREFIX[type.to_s][:prefix] + args[0][0...(MAX_TERM_LENGTH-1)]
    867     else
    868       raise "Invalid term type #{type}"
    869     end
    870   end
    871 end
    872 
    873 end
    874 
    875 class Xapian::Document
    876   def entry
    877     Marshal.load data
    878   end
    879 
    880   def entry=(x)
    881     self.data = Marshal.dump x
    882   end
    883 
    884   def index_text text, prefix, weight=1
    885     term_generator = Xapian::TermGenerator.new
    886     term_generator.stemmer = Xapian::Stem.new($config[:stem_language])
    887     term_generator.document = self
    888     term_generator.index_text text, weight, prefix
    889   end
    890 
    891   alias old_add_term add_term
    892   def add_term term
    893     if term.length <= Redwood::Index::MAX_TERM_LENGTH
    894       old_add_term term, 0
    895     else
    896       warn "dropping excessively long term #{term}"
    897     end
    898   end
    899 end