[backend] refactor & split concerns for stream

[JeremyCloarec]

Proposed changes

• added a stream client abstraction layer with stream-handler.ts and stream-utils.ts. stream-handler.ts handles the stream buisness logic independently from the stream client layer
• added a stream client interface to formalize stream client contract: all stream clients now have to implement the RawStreamClient interface defined in stream-utils.ts
• moved raw redis stream logic from redis.ts to redis-stream.ts. redis-streams.ts implements a valid RawStreamClient that can be used by stream-handler.ts

Related issues

Checklist

• [x] I consider the submitted work as finished
• [x] I tested the code for its functionality
• [ ] I wrote test cases for the relevant uses case (coverage and e2e)
• [ ] I added/update the relevant documentation (either on github or on notion)
• [ ] Where necessary I refactored code to improve the overall quality

Further comments

[codecov[bot]]

Codecov Report

:x: Patch coverage is 59.96132% with 207 lines in your changes missing coverage. Please review. :white_check_mark: Project coverage is 30.84%. Comparing base (23e924e) to head (4d2f06b). :warning: Report is 1 commits behind head on master.

Files with missing lines	Patch %	Lines
...tform/opencti-graphql/src/database/redis-stream.ts	34.57%	140 Missing :warning:
...ncti-graphql/src/database/stream/stream-handler.ts	75.93%	32 Missing :warning:
...platform/opencti-graphql/src/manager/pirManager.ts	7.69%	12 Missing :warning:
...pencti-graphql/src/database/stream/stream-utils.ts	96.00%	4 Missing :warning:
...cti-platform/opencti-graphql/src/initialization.js	0.00%	3 Missing :warning:
...orm/opencti-graphql/src/manager/activityManager.ts	0.00%	3 Missing :warning:
...cti-platform/opencti-graphql/src/database/redis.ts	81.81%	2 Missing :warning:
...tform/opencti-graphql/src/graphql/sseMiddleware.js	50.00%	2 Missing :warning:
...latform/opencti-graphql/src/manager/ruleManager.ts	50.00%	2 Missing :warning:
...rm/opencti-graphql/src/manager/fileIndexManager.ts	66.66%	1 Missing :warning:
... and 6 more

Additional details and impacted files

@@            Coverage Diff             @@
##           master   #13272      +/-   ##
==========================================
+ Coverage   30.82%   30.84%   +0.02%     
==========================================
  Files        2910     2913       +3     
  Lines      192184   192275      +91     
  Branches    39167    39174       +7     
==========================================
+ Hits        59241    59315      +74     
- Misses     132943   132960      +17     

Flag	Coverage Δ
opencti	30.84% <59.96%> (+0.02%)	:arrow_up:
opencti-front	2.47% <ø> (+<0.01%)	:arrow_up:
opencti-graphql	68.21% <59.96%> (+0.01%)	:arrow_up:

Flags with carried forward coverage won't be shown. Click here to find out more.

:umbrella: View full report in Codecov by Sentry.
:loudspeaker: Have feedback on the report? Share it here.

:rocket: New features to boost your workflow:

• :snowflake: Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• :package: JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[maelv-filigran]

~~1. Complete lack of testing for critical refactoring~~

~~Why this is a blocker:~~

~~- We are refactoring the core of the event system (creation, update, delete, merge of all entities)~~ ~~- No unit tests on the new RawStreamClient abstraction~~ ~~- No integration tests to verify that the old behavior is preserved~~ ~~- Streaming is critical: loss of events = data corruption, inconsistency between systems.~~

~~Impact:~~

~~- High risk of silent regression in production~~ ~~- Impossible to validate that the RawStreamClient contract is respected~~

~~2. Potential race condition in rawCreateStreamProcessor~~

~~Problem:~~

~~- streamListening is modified in shutdown() while processStep() is reading it~~ ~~- No mutual exclusion mechanism~~ ~~- Concurrent calls to start() or shutdown() are unprotected~~

3. Silent error handling in rawCreateStreamProcessor

Multiple issues:

1. No distinction between recoverable and fatal errors:

   • Temporary Redis network error → retry OK
   • Parsing error in streamResult → fatal, infinite loop
   • Exception in user callback → propagated or swallowed?

2. Unsafe destructuring:

   const [, results] = streamResult[0];  // If streamResult is malformed → crash
   

3. processStreamResult can throw but is handled the same as a Redis error

4. No circuit breaker: if Redis is down, we wait 5 seconds indefinitely. Maybe exponential backoff retry ?

4. Redis client created/destroyed on each call in rawFetchStreamEventsRangeFromEventId

Problem:

• Creating a Redis connection is expensive (TCP handshake, auth, select db)
• If called in a loop (paging, sync), performance is degraded
• The “client-per-request” pattern is not optimal for Redis

5. RawStreamClient interface not completely abstract

Issues:

1. rawEvent: string[] exposes the Redis XADD format:

   // Redis format: [‘field1’, ‘value1’, ‘field2’, ‘value2’]

Another implementation (Kafka) should marshal to string[] and then unmarshal

2. AuthContext and AuthUser in the interface:

   • Coupling with the OpenCTI business domain
   • A pure stream client should not know about AuthContext
   • Only stream-handler.ts should handle this mapping

3. initializeStreams optional: weak contract

[xfournet]

I'm surprised that code coverage is not very high, but probably it is not taking account end to end tests ?

[JeremyCloarec]

I'm surprised that code coverage is not very high, but probably it is not taking account end to end tests ?

I'm not sure either, but I think that it's because these method are not tested directly but indirectly, through the managers running on the test instance