Instruction file imported from Youn-Young-Seon/june-nest (.cursor/rules/fxts-guidelines.mdc). Copyright stays with the author.

FxTS Functional Programming Guidelines

Overview

This project uses FxTS for functional programming patterns in TypeScript/JavaScript. FxTS provides two main categories of functions: Lazy (for lazy evaluation and iterators) and Strict (for strict evaluation on arrays and objects).

🚨 CRITICAL: Understanding Lazy Evaluation and Iterator Protocol

⚠️ Key Concept: Iterator vs Array Returns

FxTS functions return Iterators, NOT arrays. This is crucial for avoiding type errors.

// ❌ WRONG: Assuming map returns an array
const result = pipe(
  users,
  map(user => ({ ...user, processed: true })),
  // This is an Iterator, not an array!
);

// ✅ CORRECT: Understanding Iterator flow
const result = pipe(
  users,
  map(user => ({ ...user, processed: true })),
  toArray  // Converts Iterator to Array (may return Promise<T[]>!)
);

🔑 When toArray Returns Promise<T[]>

The toArray function can return either T[] or Promise<T[]> depending on pipeline complexity:

// Simple case - returns T[]
const result = pipe([1, 2, 3], map(x => x * 2), toArray);

// Complex case - returns Promise<T[]>  
const result = pipe(
  largeDataset,
  filter(item => expensiveCheck(item)),
  map(item => transform(item)),
  toArray  // ⚠️ This may return Promise<T[]>!
);

Function Categories

Lazy Functions

Use lazy functions for:

Large datasets that don't need to be fully processed
Chaining operations that may not need all results
Memory-efficient data processing
Infinite or very large sequences

Common lazy functions:

map, filter, take, drop, chunk, flatten, uniq
range, repeat, cycle for sequence generation
concurrent, concurrentPool for async operations

Strict Functions

Use strict functions for:

Small to medium datasets
Operations that need immediate results
Final transformations in a pipeline
Simple data queries

Common strict functions:

reduce, find, some, every, includes
groupBy, countBy, partition, sortBy
pick, omit, prop, props for object manipulation

🚀 ESSENTIAL: Asynchronous Pipeline Best Practices

💡 Rule #1: Use toAsync for Complex Pipelines

When working with complex transformations or when toArray returns a Promise, always use toAsync:

// ✅ BEST PRACTICE: Explicit async pipeline
const processedUsers = await pipe(
  users,
  toAsync,  // 🔑 KEY: Convert to AsyncIterable
  filter(user => user.isActive),
  map(user => ({
    ...user,
    displayName: user.name || user.email,
    videoCount: user.Video.length
  })),
  sortBy(user => user.createdAt),
  take(10),
  toArray   // Now safely returns Promise<T[]>
);

💡 Rule #2: Always Await Async Pipelines

When using toAsync or when pipeline returns Promise, always use await:

// ❌ WRONG: Not awaiting Promise
const result = pipe(
  data,
  toAsync,
  map(item => process(item)),
  toArray,
  items => items.filter(...)  // ERROR: items is Promise, not array
);

// ✅ CORRECT: Proper await handling
const processedItems = await pipe(
  data,
  toAsync,
  map(item => process(item)),
  toArray
);
const result = processedItems.filter(...);  // Now items is actual array

💡 Rule #3: Separate Transformation and Post-Processing

For better type safety and readability:

// ✅ BEST PRACTICE: Clear separation
const transformedData = await pipe(
  rawData,
  toAsync,
  filter(isValid),
  map(transform),
  sortBy(item => item.priority),
  toArray
);

// Now use regular array methods safely
const finalResult = transformedData
  .slice(0, 10)
  .map(item => ({ ...item, processed: true }));

Pipeline Composition

Preferred Pattern: Use `pipe` for Data Transformation

// ✅ GOOD: Clear data transformation pipeline
const processUsers = (users: User[]) =>
  pipe(
    users,
    filter(user => user.isActive),
    map(user => ({ ...user, displayName: `${user.firstName} ${user.lastName}` })),
    sortBy(user => user.createdAt),
    take(10)
  );

Use `pipeLazy` for Lazy Evaluation

// ✅ GOOD: Lazy pipeline for large datasets
const processLargeDataset = (data: Iterable<Item>) =>
  pipeLazy(
    data,
    filter(item => item.isValid),
    map(item => transform(item)),
    chunk(100),
    take(5) // Only process first 5 chunks
  );

Avoid Nested Function Calls

// ❌ BAD: Hard to read nested calls
const result = take(10, 
  sortBy(user => user.createdAt,
    map(user => ({ ...user, displayName: `${user.firstName} ${user.lastName}` }),
      filter(user => user.isActive, users)
    )
  )
);

// ✅ GOOD: Use pipe for readability
const result = pipe(
  users,
  filter(user => user.isActive),
  map(user => ({ ...user, displayName: `${user.firstName} ${user.lastName}` })),
  sortBy(user => user.createdAt),
  take(10)
);

Data Transformation Best Practices

Object Manipulation

// ✅ GOOD: Use FxTS object utilities
const userSummary = pipe(
  user,
  pick(['id', 'name', 'email']),
  evolve({
    name: name => name.toUpperCase(),
    email: email => email.toLowerCase()
  })
);

// ✅ GOOD: Safe property access
const userName = prop('name', user);
const userProps = props(['name', 'email'], user);

Array Processing

// ✅ GOOD: Use appropriate functions for array operations
const activeUsers = pipe(
  users,
  filter(user => user.isActive),
  uniqBy(user => user.email), // Remove duplicates by email
  sortBy(user => user.lastName)
);

// ✅ GOOD: Use reduce for aggregations
const userStats = pipe(
  users,
  reduce((acc, user) => ({
    total: acc.total + 1,
    active: acc.active + (user.isActive ? 1 : 0)
  }), { total: 0, active: 0 })
);

Conditional Logic

// ✅ GOOD: Use functional conditional utilities
const processUser = (user: User) => pipe(
  user,
  when(user => user.isNewUser, evolve({ status: () => 'pending' })),
  unless(user => user.isVerified, omit(['sensitiveData']))
);

Performance Considerations

When to Use toAsync

Use toAsync in these critical scenarios:

Complex transformations that might involve async operations
Large datasets where lazy evaluation is beneficial
When toArray returns Promise (detected through type errors)
Service layer methods for consistent async handling

// ✅ GOOD: Complex transformation pipeline
const analytics = await pipe(
  users,
  toAsync,  // Handle complex processing
  filter(user => user.isActive),
  map(async user => ({
    ...user,
    stats: await calculateUserStats(user)  // Async operation
  })),
  sortBy(user => user.stats.score),
  take(100),
  toArray
);

Memory Efficiency with Lazy Evaluation

// ✅ GOOD: Memory-efficient processing
const topUsers = await pipe(
  millionsOfUsers,
  toAsync,
  filter(user => user.isActive),
  sortBy(user => user.score),
  take(10),  // Only process top 10, not all millions
  toArray
);

Lazy vs Strict Choice

// ✅ GOOD: Use lazy for large datasets or partial processing
const firstActiveUser = pipe(
  users, // Large array
  filter(user => user.isActive), // Lazy - stops when first match found
  take(1), // Only need first result
  toArray // Convert to array at the end
);

// ✅ GOOD: Use strict for small datasets or complete processing
const userCount = pipe(
  users, // Small array, need all results
  filter(user => user.isActive),
  size // Need complete count
);

Concurrent Processing

// ✅ GOOD: Use concurrent for async operations
const enrichedUsers = await pipe(
  users,
  toAsync,  // Always use toAsync with concurrent
  concurrent(async user => ({
    ...user,
    profile: await fetchUserProfile(user.id)
  })),
  toArray
);

// ✅ GOOD: Use concurrentPool for rate limiting
const processedUsers = await pipe(
  users,
  toAsync,  // Always use toAsync with concurrentPool
  concurrentPool(5, async user => await processUser(user)), // Max 5 concurrent
  toArray
);

Type Safety Guidelines

Generic Functions

// ✅ GOOD: Explicit type annotations for complex transformations
const transformUsers = <T extends User>(users: T[]): UserSummary[] =>
  pipe(
    users,
    map((user: T): UserSummary => ({
      id: user.id,
      name: user.name,
      isActive: user.isActive
    }))
  );

Utility Functions

// ✅ GOOD: Use FxTS type guards
const validateUser = (user: unknown): user is User => 
  isObject(user) && 
  isString(prop('name', user)) && 
  isBoolean(prop('isActive', user));

// ✅ GOOD: Use isEmpty for null/undefined checks
const hasValidEmail = (user: User): boolean => 
  !isEmpty(user.email) && isString(user.email);

Common Patterns

Data Aggregation

// ✅ GOOD: Group and aggregate data
const usersByDepartment = pipe(
  users,
  groupBy(user => user.department),
  map(users => ({
    count: size(users),
    activeCount: pipe(users, filter(user => user.isActive), size)
  }))
);

Error Handling

// ✅ GOOD: Use throwIf for validation
const validateAndProcess = (data: unknown) => pipe(
  data,
  throwIf(isEmpty, 'Data cannot be empty'),
  throwIf(data => !isArray(data), 'Data must be an array'),
  map(processItem)
);

Memoization

// ✅ GOOD: Use memoize for expensive computations
const expensiveComputation = memoize((input: string) => {
  // Expensive operation
  return computeResult(input);
});

🚨 Critical Anti-Patterns to Avoid

❌ Anti-Pattern #1: Ignoring Promise Returns

// ❌ BAD: Ignoring that toArray might return Promise
const result = pipe(data, map(...), toArray);
return result.length;  // ERROR if result is Promise

// ✅ GOOD: Handling Promise correctly
const result = await pipe(data, toAsync, map(...), toArray);
return result.length;

❌ Anti-Pattern #2: Mixing Sync/Async Inconsistently

// ❌ BAD: Inconsistent async handling
const part1 = pipe(data, map(...), toArray);  // Might be Promise
const part2 = data.filter(...);  // Always array
return [...part1, ...part2];  // ERROR if part1 is Promise

// ✅ GOOD: Consistent async handling
const part1 = await pipe(data, toAsync, map(...), toArray);
const part2 = data.filter(...);
return [...part1, ...part2];

❌ Anti-Pattern #3: Not Using toAsync When Needed

// ❌ BAD: Complex pipeline without toAsync
const result = pipe(
  complexData,
  filter(expensiveCheck),
  map(heavyTransform),
  toArray,
  items => items.slice(0, 10)  // ERROR: items might be Promise
);

// ✅ GOOD: Explicit async handling
const items = await pipe(
  complexData,
  toAsync,
  filter(expensiveCheck),
  map(heavyTransform),
  toArray
);
const result = items.slice(0, 10);

❌ Anti-Pattern #4: Mixing Paradigms

// ❌ BAD: Mixing imperative and functional styles
const processUsers = (users: User[]) => {
  const filtered = filter(user => user.isActive, users);
  const result = [];
  for (const user of filtered) {
    result.push({ ...user, processed: true });
  }
  return result;
};

// ✅ GOOD: Pure functional approach
const processUsers = async (users: User[]) => await pipe(
  users,
  toAsync,
  filter(user => user.isActive),
  map(user => ({ ...user, processed: true })),
  toArray
);

❌ Anti-Pattern #5: Ignoring Lazy Evaluation Benefits

// ❌ BAD: Converting to array too early
const result = pipe(
  largeDataset,
  toArray, // Loses lazy evaluation benefits
  filter(item => item.isValid),
  take(10)
);

// ✅ GOOD: Keep lazy until the end
const result = await pipe(
  largeDataset,
  toAsync,
  filter(item => item.isValid),
  take(10),
  toArray // Convert only when needed
);

Integration with NestJS

Service Layer Best Practices

// ✅ EXCELLENT: Proper async FxTS in service
@Injectable()
export class UserService {
  async getActiveUsers(): Promise<UserSummary[]> {
    const users = await this.prisma.user.findMany({ ... });
    
    return await pipe(
      users,
      toAsync,  // 🔑 Always use toAsync for service methods
      filter(user => user.isActive),
      map(user => this.toUserSummary(user)),
      sortBy(user => user.createdAt),
      take(10),
      toArray
    );
  }
  
  async getUserAnalytics(): Promise<UserAnalytics> {
    const users = await this.prisma.user.findMany({ ... });
    
    const [activeUsers, usersByRole, topUploaders] = await Promise.all([
      pipe(
        users,
        toAsync,
        filter(user => user.isActive),
        toArray
      ),
      pipe(
        users,
        toAsync,
        groupBy(user => user.role),
        toArray
      ),
      pipe(
        users,
        toAsync,
        sortBy(user => user.Video.length),
        take(5),
        toArray
      )
    ]);
    
    return { activeUsers, usersByRole, topUploaders };
  }
}

Controller Layer

// ✅ GOOD: Transform data in controllers
@Controller('users')
export class UserController {
  @Get()
  async getUsers(@Query() query: GetUsersQuery) {
    const users = await this.userService.findUsers(query);
    
    return await pipe(
      users,
      toAsync,  // Use toAsync in controllers too
      map(user => omit(['password', 'internalId'], user)),
      when(() => query.includeInactive, identity),
      unless(() => query.includeInactive, filter(user => user.isActive)),
      toArray
    );
  }
}

Error Handling in Services

@Injectable()
export class DataService {
  async processData(data: DataInput[]): Promise<ProcessedData[]> {
    try {
      return await pipe(
        data,
        toAsync,
        filter(this.isValid),
        map(async item => await this.transform(item)),
        toArray
      );
    } catch (error) {
      this.logger.error('FxTS pipeline failed:', error);
      throw new InternalServerErrorException('Data processing failed');
    }
  }
}

## Testing with FxTS

### Unit Tests for Async Pipelines
```typescript
// ✅ EXCELLENT: Test async FxTS pipelines
describe('UserService', () => {
  it('should process users correctly', async () => {
    const users = [
      { id: 1, name: 'John', isActive: true },
      { id: 2, name: 'Jane', isActive: false }
    ];
    
    const result = await pipe(
      users,
      toAsync,  // Use toAsync in tests too
      filter(user => user.isActive),
      map(user => user.name),
      toArray
    );
    
    expect(result).toEqual(['John']);
  });
  
  it('should handle empty input', async () => {
    const result = await pipe(
      [],
      toAsync,
      map(x => x * 2),
      toArray
    );
    
    expect(result).toEqual([]);
  });
  
  it('should handle errors properly', async () => {
    const users = [{ id: 1, name: 'John' }];
    
    await expect(
      pipe(
        users,
        toAsync,
        map(user => { throw new Error('Test error'); }),
        toArray
      )
    ).rejects.toThrow('Test error');
  });
});

Integration Tests

// ✅ GOOD: Test service integration
describe('UserService Integration', () => {
  it('should get analytics correctly', async () => {
    const service = new UserService(mockPrisma);
    
    const analytics = await service.getUserAnalytics();
    
    expect(analytics.activeUsers).toBeDefined();
    expect(analytics.usersByRole).toBeDefined();
    expect(analytics.topUploaders).toBeDefined();
  });
});

📋 Quick Reference Summary

🔑 Key Rules to Remember:

Always use toAsync for complex pipelines and service methods
Always await when using toAsync or when toArray might return Promise
Separate FxTS transformations from regular array operations
Handle type errors by understanding Iterator vs Array returns
Use lazy evaluation for memory efficiency with large datasets
Test async pipelines properly with await in tests
Add proper error handling around complex FxTS operations

🚨 Critical Pattern:

// ✅ THE GOLDEN PATTERN for NestJS Services
const result = await pipe(
  data,
  toAsync,  // Always start with toAsync
  filter(...),
  map(...),
  sortBy(...),
  take(...),
  toArray   // Always end with toArray
);

Remember: FxTS promotes immutability, composability, and clear data transformation pipelines. Always prefer functional approaches over imperative ones when working with data transformations.

🌊 Advanced Streaming & Generator Patterns

🚨 Critical Issue: `tap` Function with AsyncIterableIterator

Problem: When using tap with FxTS pipelines, you cannot directly access properties of items in an AsyncIterableIterator.

// ❌ BAD: tap on AsyncIterableIterator
const result = await pipe(
  events,
  toAsync,
  map(event => transform(event)),
  tap(event => console.log(event.type)), // ERROR: Property 'type' does not exist
  toArray
);

// ✅ GOOD: Convert to array before tap
const result = await pipe(
  events,
  toAsync,
  map(event => transform(event)),
  toArray,  // Convert to array first
  tap(events => events.forEach(event => 
    console.log(event.type) // Now we can access properties
  ))
);

🔑 Key Pattern: Logging in Streaming Pipelines

// ✅ EXCELLENT: Proper logging pattern
const uploadEvents = await pipe(
  recentUploads,
  toAsync,
  filter(video => video.uploadedBy !== null),
  map((video): VideoUploadEvent => ({
    id: `upload-${video.idx}-${Date.now()}`,
    type: 'VIDEO_UPLOAD',
    timestamp: new Date(),
    data: { /* ... */ }
  })),
  toArray, // Convert to array before logging
  tap(events => events.forEach(event => 
    this.logger.log(`📹 Video upload event: ${event.data.filename}`)
  ))
);

🌊 Generator Functions for Infinite Streams

// ✅ EXCELLENT: AsyncGenerator for infinite event streams
async *createEventStream(): AsyncGenerator<StreamEvent, never, unknown> {
  let eventId = 1;
  
  while (true) {
    try {
      // Generate events from real data sources
      const recentData = await this.fetchRecentData();
      const event = await this.generateEvent(eventId++, recentData);
      
      yield event;
      
      // Control flow rate
      await delay(Math.random() * 2000 + 500); // 0.5-2.5 seconds
      
    } catch (error) {
      // Error recovery in streams
      yield {
        id: `error-${eventId++}`,
        type: 'ERROR_EVENT',
        timestamp: new Date(),
        data: { error: error.message }
      };
      
      await delay(5000); // Longer delay on errors
    }
  }
}

🔄 Backpressure Control in Streaming

// ✅ EXCELLENT: Controlled batch processing with backpressure
async processStreamInBackground(): Promise<void> {
  try {
    let eventBuffer: StreamEvent[] = [];
    
    for await (const event of this.createEventStream()) {
      if (!this.isStreaming) break;
      
      eventBuffer.push(event);
      
      // Backpressure: Process when buffer is full
      if (eventBuffer.length >= 10) {
        await this.processBatch(eventBuffer);
        eventBuffer = []; // Clear buffer
      }
    }
  } catch (error) {
    this.logger.error('Stream processing error:', error);
    this.isStreaming = false;
  }
}

private async processBatch(events: StreamEvent[]): Promise<void> {
  await pipe(
    events,
    toAsync,
    chunk(5), // Process in smaller chunks
    map(async (eventChunk) => {
      return await pipe(
        eventChunk,
        toAsync,
        map(async (event) => {
          await delay(100); // Rate limiting
          await this.processEvent(event);
          return event;
        }),
        toArray
      );
    }),
    toArray
  );
}

📊 Real-time Window Analysis

// ✅ EXCELLENT: Time-based event window analysis
async analyzeEventWindow(events: StreamEvent[]): Promise<WindowAnalysis> {
  return await pipe(
    events,
    toAsync,
    toArray, // Convert for analysis
    
    // Analysis transformation
    (eventArray) => {
      const eventCounts = eventArray.reduce((acc, event) => {
        acc[event.type] = (acc[event.type] || 0) + 1;
        return acc;
      }, {} as Record<string, number>);
      
      const topEventType = Object.entries(eventCounts)
        .sort(([,a], [,b]) => b - a)[0]?.[0] || 'none';
      
      return {
        windowStart: new Date(Date.now() - 60000), // 1 minute window
        windowEnd: new Date(),
        eventCounts,
        totalEvents: eventArray.length,
        eventsPerSecond: eventArray.length / 60,
        topEventType,
        videoUploads: eventCounts['VIDEO_UPLOAD'] || 0,
        userActivities: eventCounts['USER_ACTIVITY'] || 0,
        systemMetrics: eventCounts['SYSTEM_METRIC'] || 0
      };
    }
  );
}

🎯 Event Stream Sampling

// ✅ EXCELLENT: Probabilistic sampling for high-volume streams
async sampleEventStream(events: StreamEvent[], sampleRate: number = 0.1): Promise<StreamEvent[]> {
  return await pipe(
    events,
    toAsync,
    filter(() => Math.random() < sampleRate), // Probabilistic sampling
    toArray,
    tap(sampledEvents => 
      this.logger.log(`🎯 Sampled ${sampledEvents.length} events from ${events.length} total`)
    )
  );
}

🔍 Multi-Criteria Stream Filtering

// ✅ EXCELLENT: Complex filtering with multiple criteria
async filterEventStream(
  events: StreamEvent[], 
  filters: {
    eventTypes?: StreamEvent['type'][];
    timeRange?: { start: Date; end: Date };
    userId?: number;
  }
): Promise<StreamEvent[]> {
  return await pipe(
    events,
    toAsync,
    
    // Event type filter
    filter(event => {
      if (filters.eventTypes && !filters.eventTypes.includes(event.type)) {
        return false;
      }
      return true;
    }),
    
    // Time range filter
    filter(event => {
      if (filters.timeRange) {
        const eventTime = event.timestamp.getTime();
        const startTime = filters.timeRange.start.getTime();
        const endTime = filters.timeRange.end.getTime();
        return eventTime >= startTime && eventTime <= endTime;
      }
      return true;
    }),
    
    // User ID filter
    filter(event => {
      if (filters.userId && event.data.userId) {
        return event.data.userId === filters.userId;
      }
      return true;
    }),
    
    toArray,
    tap(filteredEvents => 
      this.logger.log(`🔍 Filtered ${filteredEvents.length} events from ${events.length} total`)
    )
  );
}

📈 Event Trend Analysis

// ✅ EXCELLENT: Time-series trend analysis
async analyzeEventTrends(events: StreamEvent[]): Promise<TrendAnalysis> {
  return await pipe(
    events,
    toAsync,
    
    // Add time-based grouping info
    map(event => ({
      ...event,
      hour: event.timestamp.getHours(),
      minute: Math.floor(event.timestamp.getMinutes() / 10) * 10 // 10-minute buckets
    })),
    
    toArray,
    
    // Trend analysis
    (eventArray) => {
      const hourlyTrends = eventArray.reduce((acc, event) => {
        const key = `${event.hour}:${event.minute}`;
        acc[key] = (acc[key] || 0) + 1;
        return acc;
      }, {} as Record<string, number>);
      
      const typeTrends = eventArray.reduce((acc, event) => {
        acc[event.type] = (acc[event.type] || 0) + 1;
        return acc;
      }, {} as Record<string, number>);
      
      return {
        hourlyTrends,
        typeTrends,
        totalEvents: eventArray.length,
        peakHour: Object.entries(hourlyTrends)
          .sort(([,a], [,b]) => b - a)[0]?.[0] || 'none',
        mostActiveType: Object.entries(typeTrends)
          .sort(([,a], [,b]) => b - a)[0]?.[0] || 'none'
      };
    }
  );
}

🚦 Stream Rate Control Patterns

// ✅ EXCELLENT: Rate-limited stream processing
async processWithRateLimit<T>(
  items: T[],
  processor: (item: T) => Promise<void>,
  rateLimit: number = 100 // ms between items
): Promise<void> {
  await pipe(
    items,
    toAsync,
    map(async (item) => {
      await processor(item);
      await delay(rateLimit); // Rate control
      return item;
    }),
    toArray
  );
}

// ✅ EXCELLENT: Chunked processing with delays
async processLargeDataset<T>(items: T[], chunkSize: number = 10): Promise<void> {
  await pipe(
    items,
    toAsync,
    chunk(chunkSize),
    map(async (chunk) => {
      // Process chunk
      const results = await pipe(
        chunk,
        toAsync,
        map(async (item) => await this.processItem(item)),
        toArray
      );
      
      // Delay between chunks
      await delay(1000);
      return results;
    }),
    toArray
  );
}

🚨 Streaming Anti-Patterns to Avoid

❌ Anti-Pattern #6: Logging Before Array Conversion

// ❌ BAD: Trying to log AsyncIterableIterator properties
const result = await pipe(
  data,
  toAsync,
  map(transform),
  tap(item => console.log(item.property)), // ERROR: Property access on AsyncIterableIterator
  toArray
);

// ✅ GOOD: Log after converting to array
const result = await pipe(
  data,
  toAsync,
  map(transform),
  toArray,
  tap(items => items.forEach(item => console.log(item.property)))
);

❌ Anti-Pattern #7: Infinite Streams Without Control

// ❌ BAD: No rate control or error handling
async *badInfiniteStream() {
  while (true) {
    yield await fetchData(); // No delay, no error handling
  }
}

// ✅ GOOD: Controlled infinite stream
async *goodInfiniteStream() {
  while (true) {
    try {
      const data = await fetchData();
      yield data;
      await delay(1000); // Rate control
    } catch (error) {
      console.error('Stream error:', error);
      await delay(5000); // Longer delay on error
    }
  }
}

❌ Anti-Pattern #8: No Backpressure Management

// ❌ BAD: Processing without backpressure control
for await (const event of eventStream) {
  await processEvent(event); // Can overwhelm system
}

// ✅ GOOD: Batched processing with backpressure
let buffer: Event[] = [];
for await (const event of eventStream) {
  buffer.push(event);
  
  if (buffer.length >= 10) {
    await processBatch(buffer);
    buffer = [];
  }
}

📋 Streaming Patterns Summary

🔑 Critical Streaming Rules:

Always convert to array before tap when accessing item properties
Use generators for infinite streams with proper error handling and rate control
Implement backpressure in high-volume stream processing
Add delays for rate limiting to prevent system overload
Use chunking for memory efficiency when processing large datasets
Implement proper error recovery in infinite streams
Monitor and log stream health with proper metrics

🚨 Golden Streaming Pattern:

// ✅ THE GOLDEN STREAMING PATTERN
async processEventStream(): Promise<void> {
  try {
    for await (const event of this.createEventStream()) {
      const processed = await pipe(
        [event], // Convert single event to array
        toAsync,
        filter(this.isValidEvent),
        map(this.transformEvent),
        toArray,
        tap(events => this.logEvents(events))
      );
      
      await this.persistEvents(processed);
      await delay(100); // Rate control
    }
  } catch (error) {
    this.handleStreamError(error);
  }
}

Remember: Streaming with FxTS requires careful attention to type safety, rate control, and error handling. Always prioritize system stability over processing speed.

🎯 Multi-paradigm Architecture Integration

🔑 IP:OOP:FP Paradigm Unification

Based on the Generator:Iterator:LISP = IP:OOP:FP principle from multi-paradigm programming, integrate all three paradigms effectively:

// ✅ EXCELLENT: Unified multi-paradigm approach
class DataProcessor {
  // OOP: Encapsulation and state management
  private cache = new Map<string, any>();
  private memoryThreshold = 500 * 1024 * 1024; // 500MB
  
  // IP: Imperative Iterator pattern (GoF Design Pattern)
  private createIterator<T>(data: T[]): Iterator<T> {
    let index = 0;
    return {
      next(): IteratorResult<T> {
        if (index < data.length) {
          return { value: data[index++], done: false };
        }
        return { value: undefined, done: true };
      }
    };
  }
  
  // FP: LISP-style function composition
  async processData<T>(data: T[]): Promise<ProcessedData[]> {
    return await pipe(
      data,
      toAsync,
      filter(this.isValid.bind(this)),      // OOP method binding
      map(this.transform.bind(this)),       // OOP method binding
      chunk(this.getOptimalChunkSize()),    // IP procedural logic
      flatten,                              // FP pure function
      toArray                               // FP conversion
    );
  }
  
  // IP: Procedural optimization logic
  private getOptimalChunkSize(): number {
    const memoryUsage = process.memoryUsage();
    if (memoryUsage.heapUsed > this.memoryThreshold) return 50;
    if (memoryUsage.heapUsed > 100 * 1024 * 1024) return 100;
    return 200; // Default chunk size
  }
}

🚀 GoF Iterator Pattern Integration

// ✅ EXCELLENT: GoF Iterator pattern with FxTS integration
interface CustomIterator<T> {
  next(): IteratorResult<T>;
  hasNext(): boolean;
  reset(): void;
}

class PaginatedIterator<T> implements CustomIterator<T> {
  private currentPage = 0;
  private currentIndex = 0;
  private pageCache = new Map<number, T[]>();
  
  constructor(
    private fetcher: (page: number) => Promise<T[]>,
    private pageSize: number = 100
  ) {}
  
  next(): IteratorResult<T> {
    if (!this.pageCache.has(this.currentPage)) {
      // Return promise wrapper for async data
      return { value: this.fetchPage(), done: false };
    }
    
    const page = this.pageCache.get(this.currentPage)!;
    if (this.currentIndex >= page.length) {
      this.currentPage++;
      this.currentIndex = 0;
      return this.next(); // Recursive call for next page
    }
    
    return { value: page[this.currentIndex++], done: false };
  }
  
  hasNext(): boolean {
    return this.currentPage < this.getMaxPages();
  }
  
  reset(): void {
    this.currentPage = 0;
    this.currentIndex = 0;
    this.pageCache.clear();
  }
  
  private async fetchPage(): Promise<T> {
    const page = await this.fetcher(this.currentPage);
    this.pageCache.set(this.currentPage, page);
    return page[this.currentIndex++];
  }
  
  // FxTS integration
  toFxIterable(): FxIterable<T> {
    return fx(this.toIterable());
  }
  
  private *toIterable(): IterableIterator<T> {
    while (this.hasNext()) {
      const result = this.next();
      if (!result.done) {
        yield result.value;
      }
    }
  }
}

🧮 Advanced Generator Patterns

🔄 Infinite Sequence Generation with Memory Control

// ✅ EXCELLENT: Memory-efficient infinite generators
function* fibonacciGenerator(limit?: number): Generator<number, void, unknown> {
  let [a, b] = [0, 1];
  let count = 0;
  
  while (!limit || count < limit) {
    yield a;
    [a, b] = [b, a + b];
    count++;
    
    // Memory pressure check every 1000 iterations
    if (count % 1000 === 0) {
      const memUsage = process.memoryUsage();
      if (memUsage.heapUsed > 1024 * 1024 * 1024) { // 1GB limit
        console.warn('Memory pressure detected, pausing generation');
        yield* []; // Pause generation
      }
    }
  }
}

// ✅ EXCELLENT: Adaptive batch generator
function* adaptiveBatchGenerator<T>(
  source: Iterable<T>,
  initialBatchSize: number = 100
): Generator<T[], void, unknown> {
  const iterator = source[Symbol.iterator]();
  let batchSize = initialBatchSize;
  let processingTimes: number[] = [];
  
  while (true) {
    const batch: T[] = [];
    const startTime = performance.now();
    
    // Fill batch
    for (let i = 0; i < batchSize; i++) {
      const { value, done } = iterator.next();
      if (done) break;
      batch.push(value);
    }
    
    if (batch.length === 0) break;
    
    const endTime = performance.now();
    processingTimes.push(endTime - startTime);
    
    // Adaptive batch size adjustment
    if (processingTimes.length >= 5) {
      const avgTime = processingTimes.reduce((a, b) => a + b) / processingTimes.length;
      if (avgTime > 1000) { // If batch takes > 1 second
        batchSize = Math.max(10, Math.floor(batchSize * 0.8));
      } else if (avgTime < 100) { // If batch takes < 100ms
        batchSize = Math.min(1000, Math.floor(batchSize * 1.2));
      }
      processingTimes = processingTimes.slice(-5); // Keep last 5 measurements
    }
    
    yield batch;
  }
}

🎯 LISP-style Function Composition Patterns

🔗 Higher-order Function Composition

// ✅ EXCELLENT: Function composition utilities
const compose = <A, B, C>(f: (b: B) => C, g: (a: A) => B) => (a: A): C => f(g(a));
const pipe2 = <A, B, C>(g: (a: A) => B, f: (b: B) => C) => (a: A): C => f(g(a));

// ✅ EXCELLENT: Curried function patterns for reusability
const filterBy = <T>(predicate: (item: T) => boolean) => 
  (data: Iterable<T>) => filter(predicate, data);

const mapWith = <T, U>(transformer: (item: T) => U) => 
  (data: Iterable<T>) => map(transformer, data);

const takeFirst = (n: number) => 
  <T>(data: Iterable<T>) => take(n, data);

// ✅ EXCELLENT: Reusable pipeline compositions
const activeUsersPipeline = <T extends { isActive: boolean }>(users: T[]) => pipe(
  users,
  toAsync,
  filterBy((user: T) => user.isActive),
  mapWith((user: T) => ({ ...user, processedAt: new Date() })),
  takeFirst(100),
  toArray
);

const topVideosPipeline = <T extends { views: number }>(videos: T[]) => pipe(
  videos,
  toAsync,
  filterBy((video: T) => video.views > 1000),
  sortBy((video: T) => video.views),
  reverse,
  takeFirst(10),
  toArray
);

🛡️ Function Composition with Error Handling

// ✅ EXCELLENT: Safe function composition
const safeCompose = <A, B, C>(
  f: (b: B) => C,
  g: (a: A) => B,
  errorHandler: (error: Error) => C
) => (a: A): C => {
  try {
    return f(g(a));
  } catch (error) {
    return errorHandler(error as Error);
  }
};

// Usage in service methods
@Injectable()
export class DataProcessingService {
  async processUsers(users: User[]): Promise<ProcessedUser[]> {
    return await activeUsersPipeline(users);
  }
  
  async getTopVideos(videos: Video[]): Promise<TopVideo[]> {
    return await topVideosPipeline(videos);
  }
  
  async safeProcessData<T, U>(
    data: T[],
    processor: (item: T) => U
  ): Promise<U[]> {
    const safeProcessor = safeCompose(
      (item: T) => processor(item),
      (item: T) => this.validateItem(item),
      (error: Error) => {
        this.logger.error('Processing error:', error);
        return null as U;
      }
    );
    
    return await pipe(
      data,
      toAsync,
      map(safeProcessor),
      filter((item: U | null): item is U => item !== null),
      toArray
    );
  }
}

🔧 Advanced Type Safety Patterns

👻 Phantom Types for Pipeline Safety

// ✅ EXCELLENT: Phantom types for pipeline state tracking
type Validated<T> = T & { readonly __validated: true };
type Transformed<T> = T & { readonly __transformed: true };
type Enriched<T> = T & { readonly __enriched: true };

// ✅ EXCELLENT: Type-safe pipeline stages
const validateData = <T>(data: T[]): Validated<T>[] => {
  return data.map(item => ({ ...item, __validated: true as const }));
};

const transformData = <T>(data: Validated<T>[]): Transformed<Validated<T>>[] => {
  return data.map(item => ({ ...item, __transformed: true as const }));
};

const enrichData = <T>(data: Transformed<Validated<T>>[]): Enriched<Transformed<Validated<T>>>[] => {
  return data.map(item => ({ ...item, __enriched: true as const }));
};

// ✅ EXCELLENT: Type-safe pipeline with phantom types
const processDataWithTypeStates = async <T>(
  rawData: T[]
): Promise<Enriched<Transformed<Validated<T>>>[]> => {
  return await pipe(
    rawData,
    toAsync,
    map(validateData),
    map(transformData),
    map(enrichData),
    toArray
  );
};

🎖️ Conditional Type Utilities for FxTS

// ✅ EXCELLENT: Advanced type utilities
type IsIterable<T> = T extends Iterable<infer U> ? U : never;
type IsAsyncIterable<T> = T extends AsyncIterable<infer U> ? U : never;

type PipelineResult<T, F extends (...args: any[]) => any> = 
  T extends AsyncIterable<any>
    ? AsyncIterableIterator<ReturnType<F>>
    : IterableIterator<ReturnType<F>>;

// ✅ EXCELLENT: Type-safe pipeline builder
class TypedPipelineBuilder<T> {
  constructor(private source: T) {}
  
  filter<U extends IsIterable<T>>(
    predicate: (item: U) => boolean
  ): TypedPipelineBuilder<IterableIterator<U>> {
    return new TypedPipelineBuilder(
      pipe(this.source as any, filter(predicate))
    );
  }
  
  map<U extends IsIterable<T>, R>(
    mapper: (item: U) => R
  ): TypedPipelineBuilder<IterableIterator<R>> {
    return new TypedPipelineBuilder(
      pipe(this.source as any, map(mapper))
    );
  }
  
  async toArray(): Promise<IsIterable<T>[]> {
    return await pipe(this.source as any, toAsync, toArray);
  }
}

// Usage with full type safety
const typedPipeline = new TypedPipelineBuilder(users)
  .filter(user => user.isActive)
  .map(user => ({ ...user, processed: true }))
  .toArray();

🛡️ Resource Management with Lazy Evaluation

// ✅ EXCELLENT: Resource-managed generator
function* resourceManagedGenerator<T>(
  source: Iterable<T>,
  resourceFactory: () => any,
  resourceCleaner: (resource: any) => void
): Generator<T, void, unknown> {
  const resource = resourceFactory();
  
  try {
    for (const item of source) {
      yield item;
    }
  } finally {
    resourceCleaner(resource);
  }
}

// ✅ EXCELLENT: Connection pool generator
function* connectionPoolGenerator<T>(
  source: Iterable<T>,
  connectionPool: any,
  maxConnections: number = 10
): Generator<{ item: T; connection: any }, void, unknown> {
  let activeConnections = 0;
  const connections: any[] = [];
  
  try {
    for (const item of source) {
      if (activeConnections < maxConnections) {
        const connection = connectionPool.acquire();
        connections.push(connection);
        activeConnections++;
        
        yield { item, connection };
      } else {
        // Wait for connection to be available
        const connection = connections.shift();
        yield { item, connection };
        connections.push(connection);
      }
    }
  } finally {
    // Release all connections
    connections.forEach(conn => connectionPool.release(conn));
  }
}

// Usage with FxTS
const processedData = await pipe(
  resourceManagedGenerator(
    largeDataset,
    () => createDatabaseConnection(),
    (conn) => conn.close()
  ),
  toAsync,
  map(async (item) => await processWithDB(item)),
  take(10000), // Process only first 10k items
  toArray
);

🌟 Multi-paradigm Best Practices Summary

🎯 Core Principles:

Paradigm Integration: Combine IP, OOP, and FP paradigms appropriately
Memory Efficiency: Use adaptive algorithms for large dataset processing
Type Safety: Leverage phantom types and conditional types for compile-time safety
Function Composition: Apply LISP-style function composition for reusability
Performance Monitoring: Implement adaptive optimization based on runtime metrics
Resource Management: Use generators for automatic resource cleanup
Error Handling: Implement safe composition patterns with proper error recovery

🏆 Golden Multi-paradigm Pattern:

// ✅ THE ULTIMATE MULTI-PARADIGM PATTERN
class OptimizedDataProcessor<T> {
  private memoryProcessor = new MemoryAwareProcessor<T>();
  
  async processWithAllParadigms(data: T[]): Promise<ProcessedData[]> {
    // IP: Imperative logic for optimization
    const optimalBatchSize = this.calculateOptimalBatchSize(data.length);
    
    // OOP: Encapsulated state management
    const processingState = new ProcessingState();
    
    // FP: Pure function composition
    return await pipe(
      data,
      toAsync,
      filterBy(this.isValidForProcessing.bind(this)),
      mapWith(this.transformWithContext.bind(this)),
      chunk(optimalBatchSize),
      map(async (batch) => {
        processingState.incrementBatch();
        return await this.processBatch(batch);
      }),
      flatten,
      toArray
    );
  }
  
  private calculateOptimalBatchSize(dataSize: number): number {
    // IP: Procedural optimization
    const memUsage = process.memoryUsage();
    const availableMemory = os.totalmem() - memUsage.heapUsed;
    return Math.min(1000, Math.floor(availableMemory / (dataSize * 1024)));
  }
}

Remember: Multi-paradigm programming with FxTS requires thoughtful integration of imperative, object-oriented, and functional approaches. Always consider memory efficiency, type safety, and maintainability when designing complex data processing pipelines.

📊 Performance Optimization Patterns

🧠 Memory-efficient Large Dataset Processing

// ✅ EXCELLENT: Memory-aware processor with automatic optimization
class MemoryAwareProcessor<T> {
  private memoryThreshold = 500 * 1024 * 1024; // 500MB
  private processingQueue: T[] = [];
  private processedCount = 0;
  private performanceMetrics: number[] = [];
  
  async processLargeDataset(
    source: AsyncIterable<T>,
    processor: (item: T) => Promise<any>
  ): Promise<void> {
    for await (const item of source) {
      this.processingQueue.push(item);
      
      // Dynamic threshold adjustment based on performance
      if (this.processingQueue.length >= 1000 || this.shouldFlush()) {
        await this.flushQueue(processor);
      }
    }
    
    // Process remaining items
    if (this.processingQueue.length > 0) {
      await this.flushQueue(processor);
    }
  }
  
  private shouldFlush(): boolean {
    const memUsage = process.memoryUsage();
    return memUsage.heapUsed > this.memoryThreshold;
  }
  
  private async flushQueue(processor: (item: T) => Promise<any>): Promise<void> {
    const batch = this.processingQueue.splice(0);
    const startTime = performance.now();
    
    await pipe(
      batch,
      toAsync,
      map(processor),
      chunk(this.getOptimalChunkSize()),
      map(async (chunk) => {
        const results = await Promise.all(chunk);
        this.processedCount += results.length;
        return results;
      }),
      toArray
    );
    
    // Performance tracking
    const processingTime = performance.now() - startTime;
    this.performanceMetrics.push(processingTime);
    
    // Adjust memory threshold based on performance
    this.adjustMemoryThreshold();
    
    // Force garbage collection hint
    if (global.gc) {
      global.gc();
    }
  }
  
  private getOptimalChunkSize(): number {
    const avgTime = this.performanceMetrics.length > 0 
      ? this.performanceMetrics.reduce((a, b) => a + b) / this.performanceMetrics.length 
      : 100;
    
    if (avgTime > 1000) return 25;  // Slow processing
    if (avgTime < 100) return 100;  // Fast processing
    return 50; // Default
  }
  
  private adjustMemoryThreshold(): void {
    if (this.performanceMetrics.length >= 10) {
      const avgTime = this.performanceMetrics.reduce((a, b) => a + b) / this.performanceMetrics.length;
      
      if (avgTime > 2000) { // Very slow
        this.memoryThreshold = Math.max(100 * 1024 * 1024, this.memoryThreshold * 0.8);
      } else if (avgTime < 200) { // Very fast
        this.memoryThreshold = Math.min(1024 * 1024 * 1024, this.memoryThreshold * 1.2);
      }
      
      this.performanceMetrics = this.performanceMetrics.slice(-10);
    }
  }
  
  getProcessedCount(): number {
    return this.processedCount;
  }
  
  getPerformanceMetrics(): { avgTime: number; memoryThreshold: number } {
    const avgTime = this.performanceMetrics.length > 0 
      ? this.performanceMetrics.reduce((a, b) => a + b) / this.performanceMetrics.length 
      : 0;
    
    return {
      avgTime,
      memoryThreshold: this.memoryThreshold
    };
  }
}

🛡️ Resource Management with Lazy Evaluation

// ✅ EXCELLENT: Resource-managed generator
function* resourceManagedGenerator<T>(
  source: Iterable<T>,
  resourceFactory: () => any,
  resourceCleaner: (resource: any) => void
): Generator<T, void, unknown> {
  const resource = resourceFactory();
  
  try {
    for (const item of source) {
      yield item;
    }
  } finally {
    resourceCleaner(resource);
  }
}

// ✅ EXCELLENT: Connection pool generator
function* connectionPoolGenerator<T>(
  source: Iterable<T>,
  connectionPool: any,
  maxConnections: number = 10
): Generator<{ item: T; connection: any }, void, unknown> {
  let activeConnections = 0;
  const connections: any[] = [];
  
  try {
    for (const item of source) {
      if (activeConnections < maxConnections) {
        const connection = connectionPool.acquire();
        connections.push(connection);
        activeConnections++;
        
        yield { item, connection };
      } else {
        // Wait for connection to be available
        const connection = connections.shift();
        yield { item, connection };
        connections.push(connection);
      }
    }
  } finally {
    // Release all connections
    connections.forEach(conn => connectionPool.release(conn));
  }
}

// Usage with FxTS
const processedData = await pipe(
  resourceManagedGenerator(
    largeDataset,
    () => createDatabaseConnection(),
    (conn) => conn.close()
  ),
  toAsync,
  map(async (item) => await processWithDB(item)),
  take(10000), // Process only first 10k items
  toArray
);

🌟 Multi-paradigm Best Practices Summary

🎯 Core Principles:

Paradigm Integration: Combine IP, OOP, and FP paradigms appropriately
Memory Efficiency: Use adaptive algorithms for large dataset processing
Type Safety: Leverage phantom types and conditional types for compile-time safety
Function Composition: Apply LISP-style function composition for reusability
Performance Monitoring: Implement adaptive optimization based on runtime metrics
Resource Management: Use generators for automatic resource cleanup
Error Handling: Implement safe composition patterns with proper error recovery

🏆 Golden Multi-paradigm Pattern:

// ✅ THE ULTIMATE MULTI-PARADIGM PATTERN
class OptimizedDataProcessor<T> {
  private memoryProcessor = new MemoryAwareProcessor<T>();
  
  async processWithAllParadigms(data: T[]): Promise<ProcessedData[]> {
    // IP: Imperative logic for optimization
    const optimalBatchSize = this.calculateOptimalBatchSize(data.length);
    
    // OOP: Encapsulated state management
    const processingState = new ProcessingState();
    
    // FP: Pure function composition
    return await pipe(
      data,
      toAsync,
      filterBy(this.isValidForProcessing.bind(this)),
      mapWith(this.transformWithContext.bind(this)),
      chunk(optimalBatchSize),
      map(async (batch) => {
        processingState.incrementBatch();
        return await this.processBatch(batch);
      }),
      flatten,
      toArray
    );
  }
  
  private calculateOptimalBatchSize(dataSize: number): number {
    // IP: Procedural optimization
    const memUsage = process.memoryUsage();
    const availableMemory = os.totalmem() - memUsage.heapUsed;
    return Math.min(1000, Math.floor(availableMemory / (dataSize * 1024)));
  }
}

About

FxTS Functional Programming Guidelines

Overview

🚨 CRITICAL: Understanding Lazy Evaluation and Iterator Protocol

⚠️ Key Concept: Iterator vs Array Returns

🔑 When toArray Returns Promise<T[]>

Function Categories

Lazy Functions

Strict Functions

🚀 ESSENTIAL: Asynchronous Pipeline Best Practices

💡 Rule #1: Use toAsync for Complex Pipelines

💡 Rule #2: Always Await Async Pipelines

💡 Rule #3: Separate Transformation and Post-Processing

Pipeline Composition

Preferred Pattern: Use pipe for Data Transformation

Use pipeLazy for Lazy Evaluation

Avoid Nested Function Calls

Data Transformation Best Practices

Object Manipulation

Array Processing

Conditional Logic

Performance Considerations

When to Use toAsync

Memory Efficiency with Lazy Evaluation

Lazy vs Strict Choice

Concurrent Processing

Type Safety Guidelines

Generic Functions

Utility Functions

Common Patterns

Data Aggregation

Error Handling

Memoization

🚨 Critical Anti-Patterns to Avoid

❌ Anti-Pattern #1: Ignoring Promise Returns

❌ Anti-Pattern #2: Mixing Sync/Async Inconsistently

❌ Anti-Pattern #3: Not Using toAsync When Needed

❌ Anti-Pattern #4: Mixing Paradigms

❌ Anti-Pattern #5: Ignoring Lazy Evaluation Benefits

Integration with NestJS

Service Layer Best Practices

Controller Layer

Error Handling in Services

Integration Tests

📋 Quick Reference Summary

🔑 Key Rules to Remember:

🚨 Critical Pattern:

🌊 Advanced Streaming & Generator Patterns

🚨 Critical Issue: tap Function with AsyncIterableIterator

🔑 Key Pattern: Logging in Streaming Pipelines

🌊 Generator Functions for Infinite Streams

🔄 Backpressure Control in Streaming

📊 Real-time Window Analysis

🎯 Event Stream Sampling

🔍 Multi-Criteria Stream Filtering

📈 Event Trend Analysis

🚦 Stream Rate Control Patterns

🚨 Streaming Anti-Patterns to Avoid

❌ Anti-Pattern #6: Logging Before Array Conversion

❌ Anti-Pattern #7: Infinite Streams Without Control

❌ Anti-Pattern #8: No Backpressure Management

📋 Streaming Patterns Summary

🔑 Critical Streaming Rules:

🚨 Golden Streaming Pattern:

🎯 Multi-paradigm Architecture Integration

🔑 IP:OOP:FP Paradigm Unification

🚀 GoF Iterator Pattern Integration

🧮 Advanced Generator Patterns

🔄 Infinite Sequence Generation with Memory Control

🎯 LISP-style Function Composition Patterns

🔗 Higher-order Function Composition

🛡️ Function Composition with Error Handling

🔧 Advanced Type Safety Patterns

👻 Phantom Types for Pipeline Safety

🎖️ Conditional Type Utilities for FxTS

🛡️ Resource Management with Lazy Evaluation

🌟 Multi-paradigm Best Practices Summary

🎯 Core Principles:

🏆 Golden Multi-paradigm Pattern:

📊 Performance Optimization Patterns

Preferred Pattern: Use `pipe` for Data Transformation

Use `pipeLazy` for Lazy Evaluation

🚨 Critical Issue: `tap` Function with AsyncIterableIterator