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 getRoot = require('../get-root')
11
const FilterBlocked = require('../filter-blocked')
12
13
exports.manifest = {
14
  latest: 'source',
15
  roots: 'source'
16
}
17
18
exports.init = function (ssb) {
19
  // cache mostly just to avoid reading the same roots over and over again
20
  // not really big enough for multiple refresh cycles
21
  const cache = HLRU(100)
22
23
  return {
24
    latest: function ({ onlyStarted = false } = {}) {
25
      return pull(
26
        ssb.createFeedStream({ live: true, old: false }),
27
        pull.filter((msg) => {
28
          // only bump for self if this is original posting
29
          return msg.value.author !== ssb.id || !getRoot(msg)
30
        }),
31
        pull.filter(bumpFilter),
32
        LookupRoots({ ssb, cache }),
33
34
        pull.filter(msg => {
35
          if (!onlyStarted) return true
36
          const root = msg.root || msg
37
          return root.value && root.value.author === ssb.id
38
        }),
39
40
        pull.asyncMap((item, cb) => {
41
          if (onlyStarted || isParticipant(item, ssb.id)) return cb(null, item)
42
43
          const root = item.root || item
44
          threadSummary(root.key, {
45
            recentLimit: 0,
46
            readThread: ssb.patchwork.thread.read,
47
            bumpFilter
48
          }, (err, summary) => {
49
            if (err) return cb(err)
50
            if (isParticipant(summary, ssb.id)) {
51
              cb(null, item)
52
            } else {
53
              cb()
54
            }
55
          })
56
        }),
57
58
        pull.filter()
59
      )
60
    },
61
    roots: function ({ reverse, limit, resume, onlyStarted = false }) {
62
      // use resume option if specified
63
      const opts = { reverse, old: true }
64
      if (resume) {
65
        opts[reverse ? 'lt' : 'gt'] = resume
66
      }
67
68
      return pullResume.source(ssb.createFeedStream(opts), {
69
        limit,
70
        getResume: (item) => {
71
          return item && item.rts
72
        },
73
        filterMap: pull(
74
          // BUMP FILTER
75
          pull.filter(bumpFilter),
76
77
          // LOOKUP AND ADD ROOTS
78
          LookupRoots({ ssb, cache }),
79
80
          pull.filter(msg => {
81
            if (!onlyStarted) return true
82
            const root = msg.root || msg
83
            return root.value && root.value.author === ssb.id
84
          }),
85
86
          // FILTER BLOCKED (don't bump if author blocked, don't include if root author blocked)
87
          FilterBlocked([ssb.id], {
88
            isBlocking: ssb.patchwork.contacts.isBlocking,
89
            useRootAuthorBlocks: true,
90
            checkRoot: true
91
          }),
92
93
          // DON'T REPEAT THE SAME THREAD
94
          UniqueRoots(),
95
96
          // MAP ROOT ITEMS
97
          pull.map(item => {
98
            const root = item.root || item
99
            return root
100
          }),
101
102
          // RESOLVE ROOTS WITH ABOUTS
103
          ResolveAbouts({ ssb }),
104
105
          // ADD THREAD SUMMARY
106
          pull.asyncMap((item, cb) => {
107
            threadSummary(item.key, {
108
              recentLimit: 3,
109
              readThread: ssb.patchwork.thread.read,
110
              bumpFilter,
111
              pullFilter: FilterBlocked([item.value && item.value.author, ssb.id], { isBlocking: ssb.patchwork.contacts.isBlocking })
112
            }, (err, summary) => {
113
              if (err) return cb(err)
114
              cb(null, extend(item, summary))
115
            })
116
          }),
117
118
          // only threads that I've posted in
119
          pull.filter(msg => {
120
            if (onlyStarted) return true
121
            return isParticipant(msg, ssb.id)
122
          })
123
        )
124
      })
125
    }
126
  }
127
}
128
129
function isParticipant (msg, author) {
130
  if (msg.value && msg.value.author === author) return true
131
  if (msg.bumps && msg.bumps.some(bump => bump.author === author)) return true
132
}
133
134
function isAttendee (msg) {
135
  const content = msg.value && msg.value.content
136
  return (content && content.type === 'about' && content.attendee && !content.attendee.remove)
137
}
138
139
function bumpFilter (msg) {
140
  if (isAttendee(msg)) {
141
    return 'attending'
142
  } else if (msg.value.content.type === 'post') {
143
    if (getRoot(msg)) {
144
      return 'reply'
145
    } else {
146
      return 'post'
147
    }
148
  } else if (msg.value.content.type === 'about') {
149
    return 'updated'
150
  }
151
}
152