git ssb

1
const nest = require('depnest')
2
const { h, Array: MutantArray, resolve } = require('mutant')
3
const Scroller = require('mutant-scroll')
4
const pull = require('pull-stream')
5
6
const Next = require('pull-next')
7
const get = require('lodash/get')
8
const clone = require('lodash/cloneDeep')
9
10
exports.gives = nest('app.page.blogIndex')
11
12
exports.needs = nest({
13
  'app.html.blogCard': 'first',
14
  'app.html.topNav': 'first',
15
  'app.html.sideNav': 'first',
16
  'blog.sync.isBlog': 'first',
17
  'sbot.pull.stream': 'first',
18
  'sbot.obs.connection': 'first',
19
  'history.sync.push': 'first',
20
  'keys.sync.id': 'first',
21
  'message.sync.isBlocked': 'first',
22
  'translations.sync.strings': 'first',
23
  'unread.sync.isUnread': 'first'
24
})
25
26
exports.create = (api) => {
27
  var blogsCache = MutantArray()
28
29
  return nest('app.page.blogIndex', function (location) {
30
    // location here can expected to be: { page: 'blogIndex'}
31
32
    var strings = api.translations.sync.strings()
33
34
    var blogs = Scroller({
35
      classList: ['content'],
36
      prepend: api.app.html.topNav(location),
37
      streamToTop: Source({ reverse: false, live: true, old: false, limit: 20 }),
38
      streamToBottom: Source({ reverse: true, live: false, limit: 20 }),
39
      updateTop: update,
40
      updateBottom: update,
41
      store: blogsCache,
42
      render
43
    })
44
45
    return h('Page -blogIndex', {title: strings.home}, [
46
      api.app.html.sideNav(location),
47
      blogs
48
    ])
49
  })
50
51
  function Source (opts) {
52
    const commonOpts = {
53
      query: [{
54
        $filter: {
55
          value: {
56
            content: {
57
              type: 'blog'
58
            },
59
            timestamp: { $gt: 0, $lt: undefined }
60
          }
61
        }
62
      }]
63
    }
64
65
    return pull(
66
      StepperStream(
67
        (options) => api.sbot.pull.stream(sbot => sbot.query.read(options)),
68
        Object.assign(commonOpts, opts)
69
      ),
70
      pull.filter(api.blog.sync.isBlog), // isBlog or Plog?
71
      // pull.filter(msg => !msg.value.content.root), // show only root messages
72
      pull.filter(msg => !api.message.sync.isBlocked(msg)) // this is already in feed.pull.type
73
    )
74
  }
75
76
  function update (soFar, newBlog) {
77
    soFar.transaction(() => {
78
      var object = newBlog // Value(newBlog)
79
80
      const index = indexOf(soFar, (blog) => newBlog.key === resolve(blog).key)
81
      // if blog already in cache, not needed again
82
      if (index >= 0) return
83
84
      const justOlderPosition = indexOf(soFar, (msg) => newBlog.value.timestamp > resolve(msg).value.timestamp)
85
86
      if (justOlderPosition > -1) {
87
        soFar.insert(object, justOlderPosition)
88
      } else {
89
        soFar.push(object)
90
      }
91
    })
92
  }
93
94
  function render (blog) {
95
    const { recps, channel } = blog.value.content
96
    var onClick
97
    if (channel && !recps) { onClick = (ev) => api.history.sync.push(Object.assign({}, blog, { page: 'blogShow' })) }
98
    return api.app.html.blogCard(blog, { onClick })
99
  }
100
}
101
102
function indexOf (array, fn) {
103
  for (var i = 0; i < array.getLength(); i++) {
104
    if (fn(array.get(i))) {
105
      return i
106
    }
107
  }
108
  return -1
109
}
110
111
// this is needed because muxrpc doesn't do back-pressure yet
112
// this is a modified pull-next-step for ssb-query
113
function StepperStream (createStream, _opts) {
114
  var opts = clone(_opts)
115
  var last = null
116
  var count = -1
117
118
  return Next(() => {
119
    if (last) {
120
      if (count === 0) return
121
      // mix: not sure which case this ends stream for
122
      //
123
124
      var value = get(last, ['value', 'timestamp'])
125
      if (value == null) return
126
127
      if (opts.reverse) {
128
        opts.query[0].$filter.value.timestamp.$lt = value
129
      } else {
130
        opts.query[0].$filter.value.timestamp.$gt = value
131
      }
132
      last = null
133
    }
134
135
    return pull(
136
      createStream(clone(opts)),
137
      pull.through(
138
        (msg) => {
139
          count++
140
          if (!msg.sync) {
141
            last = msg
142
          }
143
        },
144
        (err) => {
145
            // retry on errors...
146
          if (err) {
147
            count = -1
148
            return count
149
          }
150
          // end stream if there were no results
151
          if (last == null) last = {}
152
        }
153
      )
154
    )
155
  })
156
}
157