How do I add autoincrementing ids to a dynamically sized observable array?

Question

I'm using callbags, but the logic is the same as with RxJS.

I'm making an observable list that can be added to and removed from. I want to have autoincrementing ids. I don't know the correct logic to attach the ids to the items.

Here's a simplified example with just add events:

import { flatten, map, merge, pipe, scan } from 'callbag-basics';
import just from 'callbag-of';
import remember from 'callbag-remember';
import subscribe from 'callbag-subscribe';

const addEvents = remember(just('add-event', 'add-event')); // remember is shareReplay(1)
const itemIds = pipe(
  addEvents,
  scan((acc) => acc + 1, 1),
);
const initialReducers = just(() => [{ id: 1, data: 'foo' }]);
const addReducers = pipe(
  addEvents,
  map(() => pipe(
    itemIds,
    map((id) => ({ id, data: 'bar' })),
  )),
  flatten,
  map((item) => (prevState) => prevState.concat(item)),
);
const reducers = merge(initialReducers, addReducers);
const states = pipe(
  reducers,
  scan((acc, reducer) => reducer(acc), null),
);

pipe(
  states,
  subscribe((state) => {
    // Expected
    // [{ id: 1, data: 'foo' }]
    // [{ id: 1, data: 'foo' }, { id: 2, data: 'bar'}]
    // [{ id: 1, data: 'foo' }, { id: 2, data: 'bar'}, { id: 3, data: 'bar'}]
    console.log(state);
    // Actual
    // [{ id: 1, data: 'foo' }]
    // [{ id: 1, data: 'foo' }, { id: 2, data: 'bar'}]
    // [{ id: 1, data: 'foo' }, { id: 2, data: 'bar'}, { id: 2, data: 'bar'}]
    // [{ id: 1, data: 'foo' }, { id: 2, data: 'bar'}, { id: 2, data: 'bar'}, { id: 3, data: 'bar'}]
  }),
);

The id starts over from the beginning for each additional item. Instead, I want the ids to match the number of items added. Essentially, the problem is attaching a running total of events to a transformation of the event data. How do I add autoincrementing ids to a dynamically sized observable array?

No need to put the answer in callbags form. I can read RxJS. Thanks.

Edit: I installed RxJS to convert the example:

const rxjs = require('rxjs');
const { map, mergeMap, scan, shareReplay } = require('rxjs/operators');

const addEvents = rxjs.of('add-event', 'add-event').pipe(
  shareReplay(1),
);
const itemIds = addEvents.pipe(
  scan((acc) => acc + 1, 1),
);
const initialReducers = rxjs.of(() => [{ id: 1, data: 'foo' }]);
const addReducers = addEvents.pipe(
  mergeMap(() => itemIds.pipe(
    map((id) => ({ id, data: 'bar' })),
  )),
  map((item) => (prevState) => prevState.concat(item)),
);
const reducers = rxjs.merge(initialReducers, addReducers);
const states = reducers.pipe(
  scan((acc, reducer) => reducer(acc), null),
);

states.subscribe((state) => {
  // Expected
  // [{ id: 1, data: 'foo' }]
  // [{ id: 1, data: 'foo' }, { id: 2, data: 'bar'}]
  // [{ id: 1, data: 'foo' }, { id: 2, data: 'bar'}, { id: 3, data: 'bar'}]
  console.log(state);
  // Actual
  // [{ id: 1, data: 'foo' }]
  // [{ id: 1, data: 'foo' }, { id: 2, data: 'bar'}]
  // [{ id: 1, data: 'foo' }, { id: 2, data: 'bar'}, { id: 2, data: 'bar'}]
  // [{ id: 1, data: 'foo' }, { id: 2, data: 'bar'}, { id: 2, data: 'bar'}, { id: 3, data: 'bar'}]
});

Answer 1

Not familiar with the callbag library, but did you try to do:

  map(() => pipe(
   itemIds,
   map((id,index) => ({ id + index, data: 'bar' })),
  )),