How to implement rolling window (key level) in Apache Beam?

I am using Apache Beam with python SDK to build data pipelines for streaming data (in my case google PubSub).

Consider this scenario: Assume that event time ~ processing time (i.e. no late events)

Now for this snapshot, I want to calculate two sets of features:

• Cumulative count per user (triggered at every event)
• number of events in the past 60 seconds (triggered at every event)

For the first case (Cumulative count per user), I have written the following code and it is working perfectly:

    with beam.Pipeline(options = options) as p:
    data = (
    p | 'read data' >> ReadFromPubSub(subscription=SUB_PATH)
      | 'add timestamp' >> beam.ParDo(AddTimestamp())
      | 'parse data' >> beam.ParDo(ParseEvent())  
    )
    
    moving_sum = (
    data 
        | 'global_window' >> beam.WindowInto(windowfn = beam.window.GlobalWindows(),
                                             trigger=beam.trigger.Repeatedly(beam.trigger.AfterCount(1)),
                                             accumulation_mode=beam.trigger.AccumulationMode.ACCUMULATING)
        | 'extract amount' >> beam.ParDo(CollectAmount())
        | 'sum' >> beam.CombinePerKey(sum)
        | 'convert to dict' >> beam.ParDo(Convert2Dict())
        | 'cumsum_to_pqsl' >> relational_db.Write(source_config=source_config,table_config=table_config2)   
    )
    

For the second set of features, I am unable to figure out a logic which will give me the number of events per user in the last 60 seconds whenever a new event is read for that user. This use case is different from fixed window and sliding window, here the window (60 seconds) is dependent on the event.

Any help on this would be really appreciated.