git ssb

1
'use strict'
2
const pull = require('pull-stream')
3
const HLRU = require('hashlru')
4
const extend = require('xtend')
5
const pullResume = require('../pull-resume')
6
const threadSummary = require('../thread-summary')
7
const LookupRoots = require('../lookup-roots')
8
const ResolveAbouts = require('../resolve-abouts')
9
const UniqueRoots = require('../unique-roots')
10
const Paramap = require('pull-paramap')
11
const FilterBlocked = require('../filter-blocked')
12
const _ = require('lodash')
13
14
exports.manifest = {
15
  latest: 'source',
16
  roots: 'source'
17
}
18
19
exports.init = function (ssb) {
20
  // cache mostly just to avoid reading the same roots over and over again
21
  // not really big enough for multiple refresh cycles
22
  const cache = HLRU(100)
23
24
  return {
25
    latest: function () {
26
      return pull(
27
        ssb.createFeedStream({
28
          private: true,
29
          live: true,
30
          old: false
31
        }),
32
        pull.filter(msg => {
33
          return msg.value && msg.value.meta && msg.value.meta.private
34
        }),
35
        pull.filter(bumpFilter),
36
        LookupRoots({ ssb, cache })
37
        // TODO: don't bump if author blocked
38
      )
39
    },
40
    roots: function ({ reverse, limit, resume }) {
41
      // use resume option if specified
42
      const opts = { reverse, old: true }
43
44
      if (reverse) {
45
        opts.query = [
46
          {
47
            $filter: {
48
              timestamp: resume ? { $lt: resume, $gt: 0 } : { $gt: 0 }
49
            }
50
          }
51
        ]
52
      } else {
53
        opts.query = [
54
          {
55
            $filter: {
56
              timestamp: resume ? { $gt: resume } : { $gt: 0 }
57
            }
58
          }
59
        ]
60
      }
61
62
      return pullResume.source(ssb.private.read(opts), {
63
        limit,
64
        getResume: (item) => {
65
          return item.timestamp
66
        },
67
        filterMap: pull(
68
          // LOOKUP AND ADD ROOTS
69
          LookupRoots({ ssb, cache }),
70
71
          // ONLY POSTS BUMP PRIVATE (currently)
72
          pull.filter(bumpFilter),
73
74
          FilterBlocked([ssb.id], {
75
            isBlocking: ssb.patchwork.contacts.isBlocking,
76
            useRootAuthorBlocks: false, // disabled in private mode
77
            checkRoot: true
78
          }),
79
80
          // DON'T REPEAT THE SAME THREAD
81
          UniqueRoots(),
82
83
          // MAP ROOT ITEMS
84
          pull.map(item => {
85
            const root = item.root || item
86
            return root
87
          }),
88
89
          // RESOLVE ROOTS WITH ABOUTS
90
          ResolveAbouts({ ssb }),
91
92
          // ADD THREAD SUMMARY
93
          Paramap((item, cb) => {
94
            threadSummary(item.key, {
95
              recentLimit: 3,
96
              readThread: ssb.patchwork.thread.read,
97
              bumpFilter,
98
              pullFilter: FilterBlocked([ssb.id], { isBlocking: ssb.patchwork.contacts.isBlocking })
99
            }, (err, summary) => {
100
              if (err) return cb(err)
101
              cb(null, extend(item, summary, {
102
                filterResult: undefined,
103
                rootBump: bumpFilter(item)
104
              }))
105
            })
106
          }, 20)
107
        )
108
      })
109
    }
110
  }
111
}
112
113
function bumpFilter (msg) {
114
  const type = _.get(msg, 'value.content.type')
115
116
  if (type === 'post') {
117
    return { type: 'reply' }
118
  }
119
}
120