Real-Time Streaming

import { StreamConfig } from 'motia'
import { z } from 'zod'
 
export const config: StreamConfig = {
  /**
   * This will be available as context.streams.petCreation in the FlowContext
   */
  name: 'petCreation',
  
  /**
   * Schema defines the structure of stream updates
   */
  schema: z.object({ 
    message: z.string()
  }),
 
  /**
   * Use default storage for the stream
   */
  baseConfig: {
    storageType: 'default',
  },
}

// steps/typescript/create-pet.step.ts
import { ApiRouteConfig, Handlers } from 'motia';
import { z } from 'zod';
import { TSStore } from './ts-store';
 
const createPetSchema = z.object({
  name: z.string().min(1, 'Name is required').trim(),
  species: z.enum(['dog', 'cat', 'bird', 'other']),
  ageMonths: z.number().int().min(0, 'Age must be a positive number'),
  weightKg: z.number().positive().optional(),
  symptoms: z.array(z.string()).optional()
});
 
export const config: ApiRouteConfig = {
  type: 'api',
  name: 'TsCreatePet',
  path: '/ts/pets',
  method: 'POST',
  emits: ['ts.pet.created', 'ts.feeding.reminder.enqueued'],
  flows: ['TsPetManagement']
};
 
export const handler: Handlers['TsCreatePet'] = async (req, { emit, logger, streams, traceId }) => {
  try {
    const validatedData = createPetSchema.parse(req.body);
 
    const pet = TSStore.create({
      name: validatedData.name,
      species: validatedData.species,
      ageMonths: validatedData.ageMonths,
      weightKg: validatedData.weightKg,
      symptoms: validatedData.symptoms
    });
 
    if (logger) {
      logger.info('🐾 Pet created', { petId: pet.id, name: pet.name, species: pet.species, status: pet.status });
    }
 
    // Create & return the initial stream record (following working pattern)
    const result = await streams.petCreation.set(traceId, 'message', { 
      message: `Pet ${pet.name} (ID: ${pet.id}) created successfully - Species: ${pet.species}, Age: ${pet.ageMonths} months, Status: ${pet.status}` 
    });
 
    if (emit) {
      await emit({
        topic: 'ts.pet.created',
        data: { petId: pet.id, event: 'pet.created', name: pet.name, species: validatedData.species, traceId }
      } as any);
 
      await emit({
        topic: 'ts.feeding.reminder.enqueued',
        data: { petId: pet.id, enqueuedAt: Date.now(), traceId }
      } as any);
    }
 
    return { 
      status: 201, 
      body: result 
    };
 
  } catch (error) {
    if (error instanceof z.ZodError) {
      return {
        status: 400,
        body: {
          message: 'Validation error',
          errors: error.errors
        }
      };
    }
 
    return {
      status: 500,
      body: { message: 'Internal server error' }
    };
  }
};

// steps/typescript/set-next-feeding-reminder.job.step.ts
import { EventConfig, Handlers } from 'motia';
import { TSStore } from './ts-store';
 
export const config = {
  type: 'event',
  name: 'TsSetNextFeedingReminder',
  description: 'Background job that sets next feeding reminder and adds welcome notes',
  subscribes: ['ts.feeding.reminder.enqueued'],
  emits: ['ts.feeding.reminder.completed'],
  flows: ['TsPetManagement']
};
 
export const handler: Handlers['TsSetNextFeedingReminder'] = async (input, { emit, logger, streams, traceId }) => {
  const { petId, enqueuedAt } = input;
 
  if (logger) {
    logger.info('🔄 Setting next feeding reminder', { petId, enqueuedAt });
  }
 
  try {
    // Calculate next feeding time (24 hours from now)
    const nextFeedingAt = Date.now() + (24 * 60 * 60 * 1000);
    
    // Fill in non-critical details and change status to in_quarantine
    const updates = {
      notes: 'Welcome to our pet store! We\'ll take great care of this pet.',
      nextFeedingAt: nextFeedingAt,
      status: 'in_quarantine' as const
    };
 
    const updatedPet = TSStore.update(petId, updates);
    
    if (!updatedPet) {
      if (logger) {
        logger.error('❌ Failed to set feeding reminder - pet not found', { petId });
      }
      return;
    }
 
    if (logger) {
      logger.info('✅ Next feeding reminder set', { 
        petId, 
        notes: updatedPet.notes?.substring(0, 50) + '...',
        nextFeedingAt: new Date(nextFeedingAt).toISOString()
      });
    }
 
    // Stream status updates using the simple pattern
    if (streams?.petCreation && traceId) {
      await streams.petCreation.set(traceId, 'message', { 
        message: `Pet ${updatedPet.name} entered quarantine period` 
      });
 
      // Check symptoms and stream appropriate updates
      if (!updatedPet.symptoms || updatedPet.symptoms.length === 0) {
        await new Promise(resolve => setTimeout(resolve, 1000));
        await streams.petCreation.set(traceId, 'message', { 
          message: `Health check passed for ${updatedPet.name} - no symptoms found` 
        });
 
        await new Promise(resolve => setTimeout(resolve, 1000));
        await streams.petCreation.set(traceId, 'message', { 
          message: `${updatedPet.name} is healthy and ready for adoption! ✅` 
        });
      } else {
        await new Promise(resolve => setTimeout(resolve, 1000));
        await streams.petCreation.set(traceId, 'message', { 
          message: `Health check failed for ${updatedPet.name} - symptoms detected: ${updatedPet.symptoms.join(', ')}` 
        });
 
        await new Promise(resolve => setTimeout(resolve, 1000));
        await streams.petCreation.set(traceId, 'message', { 
          message: `${updatedPet.name} needs medical treatment ❌` 
        });
      }
    }
 
    if (emit) {
      (emit as any)({
        topic: 'ts.feeding.reminder.completed',
        data: { 
          petId, 
          event: 'feeding.reminder.completed',
          completedAt: Date.now(),
          processingTimeMs: Date.now() - enqueuedAt
        }
      });
    }
 
  } catch (error: any) {
    if (logger) {
      logger.error('❌ Feeding reminder job error', { petId, error: error.message });
    }
  }
};

// steps/typescript/ai-profile-enrichment.step.ts
import { EventConfig, Handlers } from 'motia';
import { TSStore, PetProfile } from './ts-store';
 
export const config = {
  type: 'event',
  name: 'TsAiProfileEnrichment',
  description: 'Agentic step that enriches pet profiles using OpenAI',
  subscribes: ['ts.pet.created'],
  emits: [],
  flows: ['TsPetManagement']
};
 
export const handler: Handlers['TsAiProfileEnrichment'] = async (input, { logger, streams, traceId }) => {
  const { petId, name, species } = input;
 
  if (logger) {
    logger.info('🤖 AI Profile Enrichment started', { petId, name, species });
  }
 
  // Stream enrichment started event
  if (streams && traceId) {
    await (streams as any).petCreation.set(traceId, 'enrichment_started', { 
      message: `AI enrichment started for ${name}`
    } as any);
  }
 
  try {
    const apiKey = process.env.OPENAI_API_KEY;
    if (!apiKey) {
      throw new Error('OPENAI_API_KEY environment variable is not set');
    }
 
    const prompt = `Generate a pet profile for adoption purposes. Pet details:
- Name: ${name}
- Species: ${species}
 
Please provide a JSON response with these fields:
- bio: A warm, engaging 2-3 sentence description that would appeal to potential adopters
- breedGuess: Your best guess at the breed or breed mix (be specific but realistic)
- temperamentTags: An array of 3-5 personality traits (e.g., "friendly", "energetic", "calm")
- adopterHints: Practical advice for potential adopters (family type, living situation, care needs)
 
Keep it positive, realistic, and adoption-focused.`;
 
    const enrichmentFields = ['bio', 'breedGuess', 'temperamentTags', 'adopterHints'];
    const enrichedProfile: any = {};
 
    const response = await fetch('https://api.openai.com/v1/chat/completions', {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${apiKey}`,
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({
        model: 'gpt-3.5-turbo',
        messages: [
          {
            role: 'system',
            content: 'You are a pet adoption specialist who creates compelling, accurate pet profiles. Always respond with valid JSON only.'
          },
          {
            role: 'user',
            content: prompt
          }
        ],
        max_tokens: 500,
        temperature: 0.7,
      }),
    });
 
    if (!response.ok) {
      throw new Error(`OpenAI API error: ${response.status} ${response.statusText}`);
    }
 
    const data = await response.json();
    const aiResponse = data.choices[0]?.message?.content;
 
    if (!aiResponse) {
      throw new Error('No response from OpenAI API');
    }
 
    let profile: PetProfile;
    try {
      profile = JSON.parse(aiResponse);
    } catch (parseError) {
      profile = {
        bio: `${name} is a wonderful ${species} looking for a loving home. This pet has a unique personality and would make a great companion.`,
        breedGuess: species === 'dog' ? 'Mixed Breed' : species === 'cat' ? 'Domestic Shorthair' : 'Mixed Breed',
        temperamentTags: ['friendly', 'loving', 'loyal'],
        adopterHints: `${name} would do well in a caring home with patience and love.`
      };
      
      if (logger) {
        logger.warn('⚠️ AI response parsing failed, using fallback profile', { petId, parseError: parseError instanceof Error ? parseError.message : String(parseError) });
      }
    }
 
    const updatedPet = TSStore.updateProfile(petId, profile);
    
    if (!updatedPet) {
      throw new Error(`Pet not found: ${petId}`);
    }
 
    if (logger) {
      logger.info('✅ AI Profile Enrichment completed', { 
        petId, 
        profile: {
          bio: profile.bio.substring(0, 50) + '...',
          breedGuess: profile.breedGuess,
          temperamentTags: profile.temperamentTags,
          adopterHints: profile.adopterHints.substring(0, 50) + '...'
        }
      });
    }
 
    // Stream each field as it's processed
    for (const field of enrichmentFields) {
      await new Promise(resolve => setTimeout(resolve, 300));
      
      const value = profile[field as keyof PetProfile];
      
      if (streams && traceId) {
        await (streams as any).petCreation.set(traceId, `progress_${field}`, { 
          message: `Generated ${field} for ${name}`
        } as any);
      }
    }
 
    // Stream enrichment completed event
    if (streams && traceId) {
      await (streams as any).petCreation.set(traceId, 'completed', { 
        message: `AI enrichment completed for ${name}`
      } as any);
    }
 
  } catch (error: any) {
    if (logger) {
      logger.error('❌ AI Profile Enrichment failed', { 
        petId, 
        error: error.message 
      });
    }
 
    const fallbackProfile: PetProfile = {
      bio: `${name} is a lovely ${species} with a unique personality, ready to find their forever home.`,
      breedGuess: species === 'dog' ? 'Mixed Breed' : species === 'cat' ? 'Domestic Shorthair' : 'Mixed Breed',
      temperamentTags: ['friendly', 'adaptable'],
      adopterHints: `${name} is looking for a patient and loving family.`
    };
 
    TSStore.updateProfile(petId, fallbackProfile);
 
    // Stream fallback profile completion
    if (streams && traceId) {
      await (streams as any).petCreation.set(traceId, 'completed', { 
        message: `AI enrichment completed with fallback profile for ${name}`
      } as any);
    }
  }
};