package main

import "log"

type Message interface{}

type Syncer interface {
    Sync() error
}

type EntityId interface{}

type Entity interface {
    Id() EntityId
    Consume(Message) (Syncer, bool)
}

type Downstream chan Message

type StopChan chan struct{}

type EntityFactory func(message Message) Entity

type EntityRunner struct {
    entity     Entity
    downstream Downstream
}

type Core struct {
    sources  map[Downstream]StopChan
    mux      Downstream
    entities map[EntityId]EntityRunner
    logger   *log.Logger

    upstreamHandlerCount int
}

func (c *Core) WithSource(src Downstream) *Core {
    stopChan := make(StopChan)
    c.sources[src] = stopChan
    go func() {
        for {
            select {
            case msg := <-src:
                c.mux <- msg
            case <-stopChan:
                return
            }
        }
    }()
    return c
}

func (c *Core) RemoveSource(src Downstream) *Core {
    if s, ok := c.sources[src]; ok {
        close(s)
        delete(c.sources, src)
    }
    return c
}

func (core *Core) Run(stopChan StopChan, efactory EntityFactory) {
    eStop := make(StopChan)
    upstream := make(chan Syncer)

    for i := 0; i < core.upstreamHandlerCount; i++ {
        go func() {
            for {
                select {
                case upstreamMsg := <-upstream:
                    upstreamMsg.Sync()
                case <-eStop:
                    return
                }
            }
        }()
    }

    for {
        select {
        case msg := <-core.mux:
            entity := efactory(msg)
            id := entity.Id()
            if entityRunner, ok := core.entities[id]; ok {
                entityRunner.downstream <- msg
            } else {
                entityDownstream := make(Downstream)
                core.entities[id] = EntityRunner{entity, entityDownstream}
                go func() {
                    for {
                        select {
                        case entityMsg := <-entityDownstream:
                            if res, ok := entity.Consume(entityMsg); ok {
                                upstream <- res
                            }
                        case <-eStop:
                            return
                        }
                    }
                }()
                entityDownstream <- msg
            }
        case <-stopChan:
            close(eStop)
            core.entities = make(map[EntityId]EntityRunner)
            return
        }
    }
}