git ssb

1
var PullObv = require('pull-obv')
2
var threadReduce = require('ssb-reduce-stream')
3
var pull = require('pull-stream')
4
var Next = require('pull-next')
5
6
var nest = require('depnest')
7
8
function isObject (o) {
9
  return 'object' === typeof o
10
}
11
12
exports.gives = nest({
13
  'state.obs.threads': true,
14
  'state.obs.channel': true
15
})
16
17
exports.needs = nest({
18
  'message.sync.unbox': 'first',
19
  'sbot.pull.log': 'first',
20
  'sbot.async.get': 'first',
21
  'feed.pull.channel': 'first'
22
})
23
24
exports.create = function (api) {
25
  var threadsObs
26
27
  function createStateObs (reduce, createStream, opts, initial) {
28
    var lastTimestamp = opts.last || Date.now()
29
    var firstTimestamp = opts.first || Date.now()
30
31
    function unbox () {
32
      return pull(
33
        pull.map(function (data) {
34
          if(isObject(data.value.content)) return data
35
          return api.message.sync.unbox(data)
36
        }),
37
        pull.filter(Boolean)
38
      )
39
    }
40
41
    var obs = PullObv(
42
      reduce,
43
      pull(
44
        Next(function () {
45
          return createStream({reverse: true, limit: 500, lt: lastTimestamp})
46
        }),
47
        pull.through(function (data) {
48
          lastTimestamp = data.timestamp
49
        }),
50
        unbox()
51
      ),
52
      //value recovered from localStorage
53
      initial
54
    )
55
56
    var getting = {}
57
    obs(function (state) {
58
      var effect = state.effect
59
      if(!effect) return
60
61
      state.effect = null
62
      if(getting[effect.key]) return 
63
64
      getting[effect.key] = true
65
      api.sbot.async.get(effect.key, (err, msg) => {
66
        if (!msg) return 
67
        obs.set(reduce(obs.value, {key: effect.key, value: msg}))
68
      })
69
    })
70
71
    //stream live messages. this *should* work.
72
    //there is no back pressure on new events
73
    //only a show more on the top (currently)
74
    pull(
75
      Next(function () {
76
        return createStream({limit: 500, gt: firstTimestamp, live: true})
77
      }),
78
      pull.drain(function (data) {
79
        if(data.sync) return
80
        firstTimestamp = data.timestamp
81
        obs.set(reduce(obs.value, data))
82
      })
83
    )
84
85
    return obs
86
  }
87
88
89
  return nest({
90
  'state.obs.channel': function (channel) {
91
92
      return createStateObs(
93
        threadReduce,
94
        function (opts) {
95
          return opts.reverse ?
96
          api.feed.pull.channel(channel)(opts):
97
          pull(api.sbot.pull.log(opts), pull.filter(function (data) {
98
            if(data.sync) return false
99
            return data.value.content.channel === channel
100
          }))
101
        },
102
        {}
103
      )
104
105
    // var channelObs = PullObv(
106
    //   threadReduce,
107
    //   createChannelStream({reverse: true, limit: 1000})
108
    // )
109
110
111
  },
112
113
  'state.obs.threads': function buildThreadObs() {
114
      if(threadsObs) return threadsObs
115
116
  // DISABLE localStorage cache. mainly disabling this to make debugging the other stuff
117
  // easier. maybe re-enable this later? also, should this be for every channel too? not sure.
118
119
  //    var initial
120
  //    try { initial = JSON.parse(localStorage.threadsState) }
121
  //    catch (_) { }
122
123
      initial = {}
124
125
      threadsObs = createStateObs(threadReduce, api.sbot.pull.log, initial, {})
126
127
      threadsObs(function (threadsState) {
128
        if(threadsState.ended && threadsState.ended !== true)
129
          console.error('threadObs error:', threadsState.ended)
130
      })
131
132
  //    var timer
133
  //    //keep localStorage up to date
134
  //    threadsObs(function (threadsState) {
135
  //      if(timer) return
136
  //      timer = setTimeout(function () {
137
  //        timer = null
138
  //        threadsState.last = lastTimestamp
139
  //        console.log('save state')
140
  //        localStorage.threadsState = JSON.stringify(threadsState)
141
  //      }, 1000)
142
  //    })
143
144
      return threadsObs
145
    }
146
  })
147
}
148
149
150